diff --git a/app/main/__init__.py b/app/main/__init__.py index 2a6bff976..6d431a63a 100644 --- a/app/main/__init__.py +++ b/app/main/__init__.py @@ -29,6 +29,7 @@ from app.main.views import ( # noqa find_users, platform_admin, email_branding, + letter_branding, conversation, organisations, notifications, diff --git a/app/main/forms.py b/app/main/forms.py index 52aeffb41..4cc423401 100644 --- a/app/main/forms.py +++ b/app/main/forms.py @@ -851,6 +851,21 @@ class ServiceUpdateEmailBranding(StripWhitespaceForm): raise ValidationError('This field is required') +class SVGFileUpload(StripWhitespaceForm): + file = FileField_wtf( + 'Upload an SVG logo', + validators=[ + FileAllowed(['svg'], 'SVG Images only!'), + DataRequired(message="You need to upload a file to submit") + ] + ) + + +class ServiceLetterBrandingDetails(StripWhitespaceForm): + name = StringField('Name of brand', validators=[DataRequired()]) + domain = GovernmentDomainField('Domain') + + class PDFUploadForm(StripWhitespaceForm): file = FileField_wtf( 'Upload a letter in PDF format to check if it fits in the printable area', diff --git a/app/main/s3_client.py b/app/main/s3_client.py deleted file mode 100644 index 5fc79c6f2..000000000 --- a/app/main/s3_client.py +++ /dev/null @@ -1,144 +0,0 @@ -import uuid - -import botocore -from boto3 import resource -from flask import current_app -from notifications_utils.s3 import s3upload as utils_s3upload - -FILE_LOCATION_STRUCTURE = 'service-{}-notify/{}.csv' -TEMP_TAG = 'temp-{user_id}_' -LOGO_LOCATION_STRUCTURE = '{temp}{unique_id}-{filename}' - - -def get_s3_object(bucket_name, filename): - s3 = resource('s3') - return s3.Object(bucket_name, filename) - - -def get_csv_location(service_id, upload_id): - return ( - current_app.config['CSV_UPLOAD_BUCKET_NAME'], - FILE_LOCATION_STRUCTURE.format(service_id, upload_id), - ) - - -def get_csv_upload(service_id, upload_id): - return get_s3_object(*get_csv_location(service_id, upload_id)) - - -def delete_s3_object(filename): - bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] - get_s3_object(bucket_name, filename).delete() - - -def persist_logo(old_name, new_name): - if old_name == new_name: - return - bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] - get_s3_object(bucket_name, new_name).copy_from( - CopySource='{}/{}'.format(bucket_name, old_name)) - delete_s3_object(old_name) - - -def get_s3_objects_filter_by_prefix(prefix): - bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] - s3 = resource('s3') - return s3.Bucket(bucket_name).objects.filter(Prefix=prefix) - - -def get_temp_truncated_filename(filename, user_id): - return filename[len(TEMP_TAG.format(user_id=user_id)):] - - -def s3upload(service_id, filedata, region): - upload_id = str(uuid.uuid4()) - bucket_name, file_location = get_csv_location(service_id, upload_id) - utils_s3upload( - filedata=filedata['data'], - region=region, - bucket_name=bucket_name, - file_location=file_location, - ) - return upload_id - - -def s3download(service_id, upload_id): - contents = '' - try: - key = get_csv_upload(service_id, upload_id) - contents = key.get()['Body'].read().decode('utf-8') - except botocore.exceptions.ClientError as e: - current_app.logger.error("Unable to download s3 file {}".format( - FILE_LOCATION_STRUCTURE.format(service_id, upload_id))) - raise e - return contents - - -def get_mou(organisation_is_crown): - bucket = current_app.config['MOU_BUCKET_NAME'] - filename = 'crown.pdf' if organisation_is_crown else 'non-crown.pdf' - attachment_filename = 'GOV.UK Notify data sharing and financial agreement{}.pdf'.format( - '' if organisation_is_crown else ' (non-crown)' - ) - try: - key = get_s3_object(bucket, filename) - return { - 'filename_or_fp': key.get()['Body'], - 'attachment_filename': attachment_filename, - 'as_attachment': True, - } - except botocore.exceptions.ClientError as exception: - current_app.logger.error("Unable to download s3 file {}/{}".format( - bucket, filename - )) - raise exception - - -def upload_logo(filename, filedata, region, user_id): - upload_file_name = LOGO_LOCATION_STRUCTURE.format( - temp=TEMP_TAG.format(user_id=user_id), - unique_id=str(uuid.uuid4()), - filename=filename - ) - bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] - utils_s3upload( - filedata=filedata, - region=region, - bucket_name=bucket_name, - file_location=upload_file_name, - content_type='image/png' - ) - - return upload_file_name - - -def permanent_logo_name(filename, user_id): - if filename.startswith(TEMP_TAG.format(user_id=user_id)): - return get_temp_truncated_filename(filename=filename, user_id=user_id) - else: - return filename - - -def delete_temp_files_created_by(user_id): - for obj in get_s3_objects_filter_by_prefix(TEMP_TAG.format(user_id=user_id)): - delete_s3_object(obj.key) - - -def delete_temp_file(filename): - if not filename.startswith(TEMP_TAG[:5]): - raise ValueError('Not a temp file: {}'.format(filename)) - - delete_s3_object(filename) - - -def set_metadata_on_csv_upload(service_id, upload_id, **kwargs): - get_csv_upload( - service_id, upload_id - ).copy_from( - CopySource='{}/{}'.format(*get_csv_location(service_id, upload_id)), - ServerSideEncryption='AES256', - Metadata={ - key: str(value) for key, value in kwargs.items() - }, - MetadataDirective='REPLACE', - ) diff --git a/app/main/views/agreement.py b/app/main/views/agreement.py index b3e41dbb5..3862be01c 100644 --- a/app/main/views/agreement.py +++ b/app/main/views/agreement.py @@ -2,8 +2,8 @@ from flask import abort, render_template, request, send_file, url_for from flask_login import login_required from app.main import main -from app.main.s3_client import get_mou from app.main.views.sub_navigation_dictionaries import features_nav +from app.s3_client.s3_mou_client import get_mou from app.utils import AgreementInfo diff --git a/app/main/views/email_branding.py b/app/main/views/email_branding.py index 5a5b3f9a4..338de3c77 100644 --- a/app/main/views/email_branding.py +++ b/app/main/views/email_branding.py @@ -4,13 +4,13 @@ from flask_login import login_required from app import email_branding_client from app.main import main from app.main.forms import SearchTemplatesForm, ServiceUpdateEmailBranding -from app.main.s3_client import ( +from app.s3_client.s3_logo_client import ( TEMP_TAG, - delete_temp_file, - delete_temp_files_created_by, - permanent_logo_name, + delete_email_temp_file, + delete_email_temp_files_created_by, + permanent_email_logo_name, persist_logo, - upload_logo, + upload_email_logo, ) from app.utils import AgreementInfo, get_logo_cdn_domain, user_is_platform_admin @@ -49,7 +49,7 @@ def update_email_branding(branding_id, logo=None): if form.validate_on_submit(): if form.file.data: - upload_filename = upload_logo( + upload_filename = upload_email_logo( form.file.data.filename, form.file.data, current_app.config['AWS_REGION'], @@ -57,11 +57,11 @@ def update_email_branding(branding_id, logo=None): ) if logo and logo.startswith(TEMP_TAG.format(user_id=session['user_id'])): - delete_temp_file(logo) + delete_email_temp_file(logo) return redirect(url_for('.update_email_branding', branding_id=branding_id, logo=upload_filename)) - updated_logo_name = permanent_logo_name(logo, session["user_id"]) if logo else None + updated_logo_name = permanent_email_logo_name(logo, session["user_id"]) if logo else None email_branding_client.update_email_branding( branding_id=branding_id, @@ -76,7 +76,7 @@ def update_email_branding(branding_id, logo=None): if logo: persist_logo(logo, updated_logo_name) - delete_temp_files_created_by(session["user_id"]) + delete_email_temp_files_created_by(session["user_id"]) return redirect(url_for('.email_branding', branding_id=branding_id)) @@ -98,7 +98,7 @@ def create_email_branding(logo=None): if form.validate_on_submit(): if form.file.data: - upload_filename = upload_logo( + upload_filename = upload_email_logo( form.file.data.filename, form.file.data, current_app.config['AWS_REGION'], @@ -106,11 +106,11 @@ def create_email_branding(logo=None): ) if logo and logo.startswith(TEMP_TAG.format(user_id=session['user_id'])): - delete_temp_file(logo) + delete_email_temp_file(logo) return redirect(url_for('.create_email_branding', logo=upload_filename)) - updated_logo_name = permanent_logo_name(logo, session["user_id"]) if logo else None + updated_logo_name = permanent_email_logo_name(logo, session["user_id"]) if logo else None email_branding_client.create_email_branding( logo=updated_logo_name, @@ -124,7 +124,7 @@ def create_email_branding(logo=None): if logo: persist_logo(logo, updated_logo_name) - delete_temp_files_created_by(session["user_id"]) + delete_email_temp_files_created_by(session["user_id"]) return redirect(url_for('.email_branding')) diff --git a/app/main/views/letter_branding.py b/app/main/views/letter_branding.py new file mode 100644 index 000000000..ec6630548 --- /dev/null +++ b/app/main/views/letter_branding.py @@ -0,0 +1,116 @@ +from flask import ( + current_app, + redirect, + render_template, + request, + session, + url_for, +) +from flask_login import login_required +from notifications_python_client.errors import HTTPError +from requests import get as requests_get + +from app import letter_branding_client +from app.main import main +from app.main.forms import ServiceLetterBrandingDetails, SVGFileUpload +from app.s3_client.s3_logo_client import ( + LETTER_TEMP_TAG, + delete_letter_temp_file, + delete_letter_temp_files_created_by, + get_letter_filename_with_no_path_or_extension, + letter_filename_for_db, + permanent_letter_logo_name, + persist_logo, + upload_letter_png_logo, + upload_letter_temp_logo, +) +from app.utils import get_logo_cdn_domain, user_is_platform_admin + + +@main.route("/letter-branding/create", methods=['GET', 'POST']) +@main.route("/letter-branding/create/", methods=['GET', 'POST']) +@login_required +@user_is_platform_admin +def create_letter_branding(logo=None): + file_upload_form = SVGFileUpload() + letter_branding_details_form = ServiceLetterBrandingDetails() + + file_upload_form_submitted = file_upload_form.file.data + details_form_submitted = request.form.get('operation') == 'branding-details' + + if file_upload_form_submitted and file_upload_form.validate_on_submit(): + upload_filename = upload_letter_temp_logo( + file_upload_form.file.data.filename, + file_upload_form.file.data, + current_app.config['AWS_REGION'], + user_id=session["user_id"] + ) + + if logo and logo.startswith(LETTER_TEMP_TAG.format(user_id=session['user_id'])): + delete_letter_temp_file(logo) + + return redirect(url_for('.create_letter_branding', logo=upload_filename)) + + if details_form_submitted and letter_branding_details_form.validate_on_submit(): + if logo: + db_filename = letter_filename_for_db(logo, session['user_id']) + png_file = get_png_file_from_svg(logo) + + try: + letter_branding_client.create_letter_branding( + filename=db_filename, + name=letter_branding_details_form.name.data, + domain=letter_branding_details_form.domain.data, + ) + + upload_letter_logos(logo, db_filename, png_file, session['user_id']) + + # TODO: redirect to all letter branding page once it exists + return redirect(url_for('main.platform_admin')) + + except HTTPError as e: + if 'domain' in e.message: + letter_branding_details_form.domain.errors.append(e.message['domain'][0]) + elif 'name' in e.message: + letter_branding_details_form.name.errors.append(e.message['name'][0]) + else: + raise e + else: + # Show error on upload form if trying to submit with no logo + file_upload_form.validate() + + return render_template( + 'views/letter-branding/manage-letter-branding.html', + file_upload_form=file_upload_form, + letter_branding_details_form=letter_branding_details_form, + cdn_url=get_logo_cdn_domain(), + logo=logo + ) + + +def get_png_file_from_svg(filename): + filename_for_template_preview = get_letter_filename_with_no_path_or_extension(filename) + + template_preview_svg_endpoint = '{}/{}.svg.png'.format( + current_app.config['TEMPLATE_PREVIEW_API_HOST'], + filename_for_template_preview + ) + + response = requests_get( + template_preview_svg_endpoint, + headers={'Authorization': 'Token {}'.format(current_app.config['TEMPLATE_PREVIEW_API_KEY'])} + ) + + return response.content + + +def upload_letter_logos(old_filename, new_filename, png_file, user_id): + persist_logo(old_filename, permanent_letter_logo_name(new_filename, 'svg')) + + upload_letter_png_logo( + permanent_letter_logo_name(new_filename, 'png'), + png_file, + current_app.config['AWS_REGION'], + ) + + delete_letter_temp_files_created_by(user_id) diff --git a/app/main/views/send.py b/app/main/views/send.py index c85280185..e61dd8935 100644 --- a/app/main/views/send.py +++ b/app/main/views/send.py @@ -42,7 +42,11 @@ from app.main.forms import ( SetSenderForm, get_placeholder_form_instance, ) -from app.main.s3_client import s3download, s3upload, set_metadata_on_csv_upload +from app.s3_client.s3_csv_client import ( + s3download, + s3upload, + set_metadata_on_csv_upload, +) from app.template_previews import TemplatePreview, get_page_count_for_letter from app.utils import ( Spreadsheet, diff --git a/app/navigation.py b/app/navigation.py index 05e43e9f2..f784360fe 100644 --- a/app/navigation.py +++ b/app/navigation.py @@ -75,6 +75,7 @@ class HeaderNavigation(Navigation): 'platform-admin': { 'add_organisation', 'create_email_branding', + 'create_letter_branding', 'email_branding', 'find_users_by_email', 'live_services', @@ -405,6 +406,7 @@ class MainNavigation(Navigation): 'conversation_updates', 'cookies', 'create_email_branding', + 'create_letter_branding', 'data_retention', 'delete_template_folder', 'delivery_and_failure', @@ -583,6 +585,7 @@ class CaseworkNavigation(Navigation): 'copy_template', 'create_api_key', 'create_email_branding', + 'create_letter_branding', 'data_retention', 'delete_service_template', 'delete_template_folder', @@ -817,6 +820,7 @@ class OrgNavigation(Navigation): 'copy_template', 'create_api_key', 'create_email_branding', + 'create_letter_branding', 'data_retention', 'delete_service_template', 'delete_template_folder', diff --git a/app/notify_client/letter_branding_client.py b/app/notify_client/letter_branding_client.py index 96a0889dc..84ab0b80f 100644 --- a/app/notify_client/letter_branding_client.py +++ b/app/notify_client/letter_branding_client.py @@ -6,5 +6,13 @@ class LetterBrandingClient(NotifyAdminAPIClient): def get_letter_branding(self): return self.get(url='/dvla_organisations') + def create_letter_branding(self, filename, name, domain): + data = { + "filename": filename, + "name": name, + "domain": domain, + } + return self.post(url="/letter-branding", data=data) + letter_branding_client = LetterBrandingClient() diff --git a/app/s3_client/__init__.py b/app/s3_client/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/app/s3_client/s3_csv_client.py b/app/s3_client/s3_csv_client.py new file mode 100644 index 000000000..f9d0b5b9f --- /dev/null +++ b/app/s3_client/s3_csv_client.py @@ -0,0 +1,57 @@ +import uuid + +import botocore +from flask import current_app +from notifications_utils.s3 import s3upload as utils_s3upload + +from app.s3_client.s3_logo_client import get_s3_object + +FILE_LOCATION_STRUCTURE = 'service-{}-notify/{}.csv' + + +def get_csv_location(service_id, upload_id): + return ( + current_app.config['CSV_UPLOAD_BUCKET_NAME'], + FILE_LOCATION_STRUCTURE.format(service_id, upload_id), + ) + + +def get_csv_upload(service_id, upload_id): + return get_s3_object(*get_csv_location(service_id, upload_id)) + + +def s3upload(service_id, filedata, region): + upload_id = str(uuid.uuid4()) + bucket_name, file_location = get_csv_location(service_id, upload_id) + utils_s3upload( + filedata=filedata['data'], + region=region, + bucket_name=bucket_name, + file_location=file_location, + ) + return upload_id + + +def s3download(service_id, upload_id): + contents = '' + try: + key = get_csv_upload(service_id, upload_id) + contents = key.get()['Body'].read().decode('utf-8') + except botocore.exceptions.ClientError as e: + current_app.logger.error("Unable to download s3 file {}".format( + FILE_LOCATION_STRUCTURE.format(service_id, upload_id))) + raise e + return contents + + +def set_metadata_on_csv_upload(service_id, upload_id, **kwargs): + get_csv_upload( + service_id, upload_id + ).copy_from( + CopySource='{}/{}'.format(*get_csv_location(service_id, upload_id)), + ServerSideEncryption='AES256', + Metadata={ + key: str(value) for key, value in kwargs.items() + }, + MetadataDirective='REPLACE', + ) diff --git a/app/s3_client/s3_logo_client.py b/app/s3_client/s3_logo_client.py new file mode 100644 index 000000000..bb8626bae --- /dev/null +++ b/app/s3_client/s3_logo_client.py @@ -0,0 +1,135 @@ +import uuid + +from boto3 import resource +from flask import current_app +from notifications_utils.s3 import s3upload as utils_s3upload + +TEMP_TAG = 'temp-{user_id}_' +EMAIL_LOGO_LOCATION_STRUCTURE = '{temp}{unique_id}-{filename}' +LETTER_PREFIX = 'letters/static/images/letter-template/' +LETTER_TEMP_TAG = LETTER_PREFIX + TEMP_TAG +LETTER_TEMP_LOGO_LOCATION = 'letters/static/images/letter-template/temp-{user_id}_{unique_id}-{filename}' + + +def get_s3_object(bucket_name, filename): + s3 = resource('s3') + return s3.Object(bucket_name, filename) + + +def delete_s3_object(filename): + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + get_s3_object(bucket_name, filename).delete() + + +def persist_logo(old_name, new_name): + if old_name == new_name: + return + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + get_s3_object(bucket_name, new_name).copy_from( + CopySource='{}/{}'.format(bucket_name, old_name)) + delete_s3_object(old_name) + + +def get_s3_objects_filter_by_prefix(prefix): + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + s3 = resource('s3') + return s3.Bucket(bucket_name).objects.filter(Prefix=prefix) + + +def get_temp_truncated_filename(filename, user_id): + return filename[len(TEMP_TAG.format(user_id=user_id)):] + + +def get_letter_filename_with_no_path_or_extension(filename): + return filename[len(LETTER_PREFIX):-4] + + +def upload_email_logo(filename, filedata, region, user_id): + upload_file_name = EMAIL_LOGO_LOCATION_STRUCTURE.format( + temp=TEMP_TAG.format(user_id=user_id), + unique_id=str(uuid.uuid4()), + filename=filename + ) + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + utils_s3upload( + filedata=filedata, + region=region, + bucket_name=bucket_name, + file_location=upload_file_name, + content_type='image/png' + ) + + return upload_file_name + + +def upload_letter_temp_logo(filename, filedata, region, user_id): + upload_filename = LETTER_TEMP_LOGO_LOCATION.format( + user_id=user_id, + unique_id=str(uuid.uuid4()), + filename=filename + ) + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + utils_s3upload( + filedata=filedata, + region=region, + bucket_name=bucket_name, + file_location=upload_filename, + content_type='image/svg+xml' + ) + + return upload_filename + + +def upload_letter_png_logo(filename, filedata, region): + bucket_name = current_app.config['LOGO_UPLOAD_BUCKET_NAME'] + utils_s3upload( + filedata=filedata, + region=region, + bucket_name=bucket_name, + file_location=filename, + content_type='image/png' + ) + + +def permanent_email_logo_name(filename, user_id): + if filename.startswith(TEMP_TAG.format(user_id=user_id)): + return get_temp_truncated_filename(filename=filename, user_id=user_id) + else: + return filename + + +def permanent_letter_logo_name(filename, extension): + return LETTER_PREFIX + filename + '.' + extension + + +def letter_filename_for_db(filename, user_id): + filename = get_letter_filename_with_no_path_or_extension(filename) + + if filename.startswith(TEMP_TAG.format(user_id=user_id)): + filename = get_temp_truncated_filename(filename=filename, user_id=user_id) + + return filename + + +def delete_email_temp_files_created_by(user_id): + for obj in get_s3_objects_filter_by_prefix(TEMP_TAG.format(user_id=user_id)): + delete_s3_object(obj.key) + + +def delete_letter_temp_files_created_by(user_id): + for obj in get_s3_objects_filter_by_prefix(LETTER_TEMP_TAG.format(user_id=user_id)): + delete_s3_object(obj.key) + + +def delete_email_temp_file(filename): + if not filename.startswith(TEMP_TAG[:5]): + raise ValueError('Not a temp file: {}'.format(filename)) + + delete_s3_object(filename) + + +def delete_letter_temp_file(filename): + if not filename.startswith(LETTER_TEMP_TAG[:43]): + raise ValueError('Not a temp file: {}'.format(filename)) + + delete_s3_object(filename) diff --git a/app/s3_client/s3_mou_client.py b/app/s3_client/s3_mou_client.py new file mode 100644 index 000000000..d86e7b2cc --- /dev/null +++ b/app/s3_client/s3_mou_client.py @@ -0,0 +1,24 @@ +import botocore +from flask import current_app + +from app.s3_client.s3_logo_client import get_s3_object + + +def get_mou(organisation_is_crown): + bucket = current_app.config['MOU_BUCKET_NAME'] + filename = 'crown.pdf' if organisation_is_crown else 'non-crown.pdf' + attachment_filename = 'GOV.UK Notify data sharing and financial agreement{}.pdf'.format( + '' if organisation_is_crown else ' (non-crown)' + ) + try: + key = get_s3_object(bucket, filename) + return { + 'filename_or_fp': key.get()['Body'], + 'attachment_filename': attachment_filename, + 'as_attachment': True, + } + except botocore.exceptions.ClientError as exception: + current_app.logger.error("Unable to download s3 file {}/{}".format( + bucket, filename + )) + raise exception diff --git a/app/templates/views/letter-branding/manage-letter-branding.html b/app/templates/views/letter-branding/manage-letter-branding.html new file mode 100644 index 000000000..060054208 --- /dev/null +++ b/app/templates/views/letter-branding/manage-letter-branding.html @@ -0,0 +1,37 @@ +{% extends "views/platform-admin/_base_template.html" %} +{% from "components/file-upload.html" import file_upload %} +{% from "components/page-footer.html" import page_footer %} +{% from "components/textbox.html" import textbox %} +{% from "components/form.html" import form_wrapper %} + +{% block per_page_title %} + Create letter branding +{% endblock %} + +{% block platform_admin_content %} + +

Add letter branding

+
+
+ {% if logo %} +
+ +
+ {% endif %} + {{ file_upload(file_upload_form.file) }} + {% call form_wrapper() %} +
+
{{textbox(letter_branding_details_form.name)}}
+
{{textbox(letter_branding_details_form.domain)}}
+ {{ page_footer( + 'Save', + button_name='operation', + button_value='branding-details', + back_link=url_for('main.platform_admin'), + back_link_text='Back to letter branding selection', + ) }} +
+ {% endcall %} +
+ +{% endblock %} diff --git a/app/utils.py b/app/utils.py index 0f7ee1134..c7ed47090 100644 --- a/app/utils.py +++ b/app/utils.py @@ -126,7 +126,7 @@ def get_errors_for_csv(recipients, template_type): def generate_notifications_csv(**kwargs): from app import notification_api_client - from app.main.s3_client import s3download + from app.s3_client.s3_csv_client import s3download if 'page' not in kwargs: kwargs['page'] = 1 diff --git a/tests/app/main/test_s3_client.py b/tests/app/main/test_s3_client.py deleted file mode 100644 index 20c6d80a3..000000000 --- a/tests/app/main/test_s3_client.py +++ /dev/null @@ -1,130 +0,0 @@ -from collections import namedtuple -from unittest.mock import Mock, call - -import pytest - -from app.main.s3_client import ( - LOGO_LOCATION_STRUCTURE, - TEMP_TAG, - delete_temp_file, - delete_temp_files_created_by, - permanent_logo_name, - persist_logo, - set_metadata_on_csv_upload, - upload_logo, -) - -bucket = 'test_bucket' -data = {'data': 'some_data'} -filename = 'test.png' -upload_id = 'test_uuid' -region = 'eu-west1' - - -@pytest.fixture -def upload_filename(fake_uuid): - return LOGO_LOCATION_STRUCTURE.format( - temp=TEMP_TAG.format(user_id=fake_uuid), unique_id=upload_id, filename=filename) - - -def test_upload_logo_calls_correct_args(client, mocker, fake_uuid, upload_filename): - mocker.patch('uuid.uuid4', return_value=upload_id) - mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) - mocked_s3_upload = mocker.patch('app.main.s3_client.utils_s3upload') - - upload_logo(filename=filename, user_id=fake_uuid, filedata=data, region=region) - - assert mocked_s3_upload.called_once_with( - filedata=data, - region=region, - file_location=upload_filename, - bucket_name=bucket - ) - - -def test_persist_logo(client, mocker, fake_uuid, upload_filename): - mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) - mocked_get_s3_object = mocker.patch('app.main.s3_client.get_s3_object') - mocked_delete_s3_object = mocker.patch('app.main.s3_client.delete_s3_object') - - new_filename = permanent_logo_name(upload_filename, fake_uuid) - - persist_logo(upload_filename, new_filename) - - assert mocked_get_s3_object.called_once_with(bucket, new_filename) - assert mocked_delete_s3_object.called_once_with(bucket, upload_filename) - - -def test_persist_logo_returns_if_not_temp(client, mocker, fake_uuid): - filename = 'logo.png' - persist_logo(filename, filename) - - mocked_get_s3_object = mocker.patch('app.main.s3_client.get_s3_object') - mocked_delete_s3_object = mocker.patch('app.main.s3_client.delete_s3_object') - - mocked_get_s3_object.assert_not_called() - mocked_delete_s3_object.assert_not_called() - - -def test_permanent_logo_name_removes_temp_tag_from_filename(upload_filename, fake_uuid): - new_name = permanent_logo_name(upload_filename, fake_uuid) - - assert new_name == 'test_uuid-test.png' - - -def test_permanent_logo_name_does_not_change_filenames_with_no_temp_tag(): - filename = 'logo.png' - new_name = permanent_logo_name(filename, filename) - - assert new_name == filename - - -def test_delete_temp_files_created_by_user(client, mocker, fake_uuid): - obj = namedtuple("obj", ["key"]) - objs = [obj(key='test1'), obj(key='test2')] - - mocker.patch('app.main.s3_client.get_s3_objects_filter_by_prefix', return_value=objs) - mocked_delete_s3_object = mocker.patch('app.main.s3_client.delete_s3_object') - - delete_temp_files_created_by(fake_uuid) - - assert mocked_delete_s3_object.called_with_args(objs[0].key) - for index, arg in enumerate(mocked_delete_s3_object.call_args_list): - assert arg == call(objs[index].key) - - -def test_delete_single_temp_file(client, mocker, fake_uuid, upload_filename): - mocked_delete_s3_object = mocker.patch('app.main.s3_client.delete_s3_object') - - delete_temp_file(upload_filename) - - assert mocked_delete_s3_object.called_with_args(upload_filename) - - -def test_does_not_delete_non_temp_file(client, mocker, fake_uuid): - filename = 'logo.png' - mocked_delete_s3_object = mocker.patch('app.main.s3_client.delete_s3_object') - - with pytest.raises(ValueError) as error: - delete_temp_file(filename) - - assert mocked_delete_s3_object.called_with_args(filename) - assert str(error.value) == 'Not a temp file: {}'.format(filename) - - -def test_sets_metadata(client, mocker): - mocked_s3_object = Mock() - mocked_get_s3_object = mocker.patch( - 'app.main.s3_client.get_csv_upload', - return_value=mocked_s3_object, - ) - - set_metadata_on_csv_upload('1234', '5678', foo='bar', baz=True) - - mocked_get_s3_object.assert_called_once_with('1234', '5678') - mocked_s3_object.copy_from.assert_called_once_with( - CopySource='test-notifications-csv-upload/service-1234-notify/5678.csv', - Metadata={'baz': 'True', 'foo': 'bar'}, - MetadataDirective='REPLACE', - ServerSideEncryption='AES256', - ) diff --git a/tests/app/main/views/test_agreement.py b/tests/app/main/views/test_agreement.py index 02bc6b34e..8da0fd3e2 100644 --- a/tests/app/main/views/test_agreement.py +++ b/tests/app/main/views/test_agreement.py @@ -78,7 +78,7 @@ def test_downloading_agreement( expected_file_served, ): mock_get_s3_object = mocker.patch( - 'app.main.s3_client.get_s3_object', + 'app.s3_client.s3_mou_client.get_s3_object', return_value=_MockS3Object(b'foo') ) user = active_user_with_permissions(fake_uuid) @@ -100,7 +100,7 @@ def test_agreement_cant_be_downloaded_unknown_crown_status( fake_uuid, ): mock_get_s3_object = mocker.patch( - 'app.main.s3_client.get_s3_object', + 'app.s3_client.s3_mou_client.get_s3_object', return_value=_MockS3Object() ) user = active_user_with_permissions(fake_uuid) @@ -116,7 +116,7 @@ def test_agreement_requires_login( mocker, ): mock_get_s3_object = mocker.patch( - 'app.main.s3_client.get_s3_object', + 'app.s3_client.s3_mou_client.get_s3_object', return_value=_MockS3Object() ) response = client.get(url_for('main.download_agreement')) @@ -142,7 +142,7 @@ def test_show_public_agreement_page( expected_status, ): mocker.patch( - 'app.main.s3_client.get_s3_object', + 'app.s3_client.s3_mou_client.get_s3_object', return_value=_MockS3Object() ) response = client.get(url_for( diff --git a/tests/app/main/views/test_email_branding.py b/tests/app/main/views/test_email_branding.py index 6b8d1f318..be702a3bd 100644 --- a/tests/app/main/views/test_email_branding.py +++ b/tests/app/main/views/test_email_branding.py @@ -6,7 +6,7 @@ from bs4 import BeautifulSoup from flask import url_for from notifications_python_client.errors import HTTPError -from app.main.s3_client import LOGO_LOCATION_STRUCTURE, TEMP_TAG +from app.s3_client.s3_logo_client import EMAIL_LOGO_LOCATION_STRUCTURE, TEMP_TAG from tests.conftest import ( mock_get_email_branding, normalize_spaces, @@ -114,7 +114,7 @@ def test_create_new_email_branding_without_logo( } mock_persist = mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') logged_in_platform_admin_client.post( url_for('.create_email_branding'), @@ -141,7 +141,7 @@ def test_cant_create_new_email_branding_with_unknown_domain( mock_create_email_branding ): mock_persist = mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') client_request.login(platform_admin_user(fake_uuid)) page = client_request.post( @@ -197,7 +197,7 @@ def test_rejects_non_canonical_domain_when_adding_email_branding( expected_error, ): mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') data = { 'logo': None, 'colour': '#ff0000', @@ -225,7 +225,7 @@ def test_create_email_branding_requires_a_name_when_submitting_logo_details( mock_create_email_branding, ): mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') data = { 'operation': 'email-branding-details', 'logo': '', @@ -252,7 +252,7 @@ def test_create_email_branding_does_not_require_a_name_when_uploading_a_file( mocker, fake_uuid, ): - mocker.patch('app.main.views.email_branding.upload_logo', return_value='temp_filename') + mocker.patch('app.main.views.email_branding.upload_email_logo', return_value='temp_filename') data = { 'file': (BytesIO(''.encode('utf-8')), 'test.png'), 'colour': '', @@ -290,14 +290,14 @@ def test_create_new_email_branding_when_branding_saved( 'brand_type': 'org_banner' } - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename=data['logo'] ) mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') logged_in_platform_admin_client.post( url_for('.create_email_branding', logo=temp_filename), @@ -342,24 +342,24 @@ def test_deletes_previous_temp_logo_after_uploading_logo( with logged_in_platform_admin_client.session_transaction() as session: user_id = session["user_id"] - temp_old_filename = LOGO_LOCATION_STRUCTURE.format( + temp_old_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename='old_test.png' ) - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename='test.png' ) - mocked_upload_logo = mocker.patch( - 'app.main.views.email_branding.upload_logo', + mocked_upload_email_logo = mocker.patch( + 'app.main.views.email_branding.upload_email_logo', return_value=temp_filename ) - mocked_delete_temp_file = mocker.patch('app.main.views.email_branding.delete_temp_file') + mocked_delete_email_temp_file = mocker.patch('app.main.views.email_branding.delete_email_temp_file') logged_in_platform_admin_client.post( url_for('main.create_email_branding', logo=temp_old_filename, branding_id=fake_uuid), @@ -367,9 +367,9 @@ def test_deletes_previous_temp_logo_after_uploading_logo( content_type='multipart/form-data' ) - assert mocked_upload_logo.called - assert mocked_delete_temp_file.called - assert mocked_delete_temp_file.call_args == call(temp_old_filename) + assert mocked_upload_email_logo.called + assert mocked_delete_email_temp_file.called + assert mocked_delete_email_temp_file.call_args == call(temp_old_filename) def test_update_existing_branding( @@ -391,14 +391,14 @@ def test_update_existing_branding( 'brand_type': 'both' } - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename=data['logo'] ) mocker.patch('app.main.views.email_branding.persist_logo') - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') logged_in_platform_admin_client.post( url_for('.update_email_branding', logo=temp_filename, branding_id=fake_uuid), @@ -431,14 +431,14 @@ def test_temp_logo_is_shown_after_uploading_logo( with logged_in_platform_admin_client.session_transaction() as session: user_id = session["user_id"] - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename='test.png' ) - mocker.patch('app.main.views.email_branding.upload_logo', return_value=temp_filename) - mocker.patch('app.main.views.email_branding.delete_temp_file') + mocker.patch('app.main.views.email_branding.upload_email_logo', return_value=temp_filename) + mocker.patch('app.main.views.email_branding.delete_email_temp_file') response = logged_in_platform_admin_client.post( url_for('main.create_email_branding'), @@ -463,12 +463,12 @@ def test_logo_persisted_when_organisation_saved( with logged_in_platform_admin_client.session_transaction() as session: user_id = session["user_id"] - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename='test.png') - mocked_upload_logo = mocker.patch('app.main.views.email_branding.upload_logo') + mocked_upload_email_logo = mocker.patch('app.main.views.email_branding.upload_email_logo') mocked_persist_logo = mocker.patch('app.main.views.email_branding.persist_logo') - mocked_delete_temp_files_by = mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocked_delete_email_temp_files_by = mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') resp = logged_in_platform_admin_client.post( url_for('.create_email_branding', logo=temp_filename), @@ -476,10 +476,10 @@ def test_logo_persisted_when_organisation_saved( ) assert resp.status_code == 302 - assert not mocked_upload_logo.called + assert not mocked_upload_email_logo.called assert mocked_persist_logo.called - assert mocked_delete_temp_files_by.called - assert mocked_delete_temp_files_by.call_args == call(user_id) + assert mocked_delete_email_temp_files_by.called + assert mocked_delete_email_temp_files_by.call_args == call(user_id) assert mock_create_email_branding.called @@ -492,11 +492,11 @@ def test_logo_does_not_get_persisted_if_updating_email_branding_client_throws_an with logged_in_platform_admin_client.session_transaction() as session: user_id = session["user_id"] - temp_filename = LOGO_LOCATION_STRUCTURE.format( + temp_filename = EMAIL_LOGO_LOCATION_STRUCTURE.format( temp=TEMP_TAG.format(user_id=user_id), unique_id=fake_uuid, filename='test.png') mocked_persist_logo = mocker.patch('app.main.views.email_branding.persist_logo') - mocked_delete_temp_files_by = mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocked_delete_email_temp_files_by = mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') mocker.patch('app.main.views.email_branding.email_branding_client.create_email_branding', side_effect=HTTPError()) logged_in_platform_admin_client.post( @@ -505,7 +505,7 @@ def test_logo_does_not_get_persisted_if_updating_email_branding_client_throws_an ) assert not mocked_persist_logo.called - assert not mocked_delete_temp_files_by.called + assert not mocked_delete_email_temp_files_by.called @pytest.mark.parametrize('colour_hex, expected_status_code', [ @@ -530,7 +530,7 @@ def test_colour_regex_validation( 'brand_type': 'org' } - mocker.patch('app.main.views.email_branding.delete_temp_files_created_by') + mocker.patch('app.main.views.email_branding.delete_email_temp_files_created_by') response = logged_in_platform_admin_client.post( url_for('.create_email_branding'), diff --git a/tests/app/main/views/test_letter_branding.py b/tests/app/main/views/test_letter_branding.py new file mode 100644 index 000000000..21eab361e --- /dev/null +++ b/tests/app/main/views/test_letter_branding.py @@ -0,0 +1,274 @@ +from io import BytesIO +from unittest.mock import Mock + +import pytest +from bs4 import BeautifulSoup +from flask import current_app, url_for +from notifications_python_client.errors import HTTPError + +from app.main.views.letter_branding import get_png_file_from_svg +from app.s3_client.s3_logo_client import LETTER_TEMP_LOGO_LOCATION + + +def test_create_letter_branding_does_not_show_branding_info(logged_in_platform_admin_client): + response = logged_in_platform_admin_client.get( + url_for('.create_letter_branding') + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert page.select_one('#logo-img > img') is None + assert page.select_one('#name').attrs.get('value') == '' + assert page.select_one('#domain').attrs.get('value') == '' + + +def test_create_letter_branding_when_uploading_valid_file( + mocker, + logged_in_platform_admin_client, + fake_uuid +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + filename = 'test.svg' + expected_temp_filename = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename=filename) + + mock_s3_upload = mocker.patch('app.s3_client.s3_logo_client.utils_s3upload') + mocker.patch('app.s3_client.s3_logo_client.uuid.uuid4', return_value=fake_uuid) + mock_delete_temp_files = mocker.patch('app.main.views.letter_branding.delete_letter_temp_file') + + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding'), + data={'file': (BytesIO(''.encode('utf-8')), filename)}, + follow_redirects=True + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert page.select_one('#logo-img > img').attrs['src'].endswith(expected_temp_filename) + assert mock_s3_upload.called + mock_delete_temp_files.assert_not_called() + + +def test_create_letter_branding_when_uploading_invalid_file(logged_in_platform_admin_client): + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding'), + data={'file': (BytesIO(''.encode('utf-8')), 'test.png')}, + follow_redirects=True + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert page.find('h1').text == 'Add letter branding' + assert page.select_one('.error-message').text.strip() == 'SVG Images only!' + + +def test_create_letter_branding_deletes_temp_files_when_uploading_a_new_file( + mocker, + logged_in_platform_admin_client, + fake_uuid, +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + temp_logo = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename='temp.svg') + + mock_s3_upload = mocker.patch('app.s3_client.s3_logo_client.utils_s3upload') + mock_delete_temp_files = mocker.patch('app.main.views.letter_branding.delete_letter_temp_file') + + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding', logo=temp_logo), + data={'file': (BytesIO(''.encode('utf-8')), 'new.svg')}, + follow_redirects=True + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert mock_s3_upload.called + assert mock_delete_temp_files.called + assert page.find('h1').text == 'Add letter branding' + + +def test_create_new_letter_branding_shows_preview_of_logo( + mocker, + logged_in_platform_admin_client, + fake_uuid +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + temp_logo = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename='temp.svg') + + response = logged_in_platform_admin_client.get( + url_for('.create_letter_branding', logo=temp_logo) + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert page.find('h1').text == 'Add letter branding' + assert page.select_one('#logo-img > img').attrs['src'].endswith(temp_logo) + + +def test_create_letter_branding_shows_an_error_when_submitting_details_with_no_logo( + logged_in_platform_admin_client, + fake_uuid +): + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding'), + data={ + 'name': 'Test brand', + 'domain': 'bl.uk', + 'operation': 'branding-details' + } + ) + + assert response.status_code == 200 + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + + assert page.find('h1').text == 'Add letter branding' + assert page.select_one('.error-message').text.strip() == 'You need to upload a file to submit' + + +@pytest.mark.parametrize('domain', ['bl.uk', '']) +def test_create_letter_branding_persists_logo_when_all_data_is_valid( + mocker, + logged_in_platform_admin_client, + fake_uuid, + domain +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + temp_logo = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename='test.svg') + + mock_letter_client = mocker.patch('app.main.views.letter_branding.letter_branding_client') + mock_template_preview = mocker.patch( + 'app.main.views.letter_branding.get_png_file_from_svg', + return_value='fake_png') + mock_persist_logo = mocker.patch('app.main.views.letter_branding.persist_logo') + mock_upload_png = mocker.patch('app.main.views.letter_branding.upload_letter_png_logo') + mock_delete_temp_files = mocker.patch('app.main.views.letter_branding.delete_letter_temp_files_created_by') + + # TODO: remove platform admin page mocks once we no longer redirect there + mocker.patch('app.main.views.platform_admin.make_columns') + mocker.patch('app.main.views.platform_admin.platform_stats_api_client.get_aggregate_platform_stats') + mocker.patch('app.main.views.platform_admin.complaint_api_client.get_complaint_count') + + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding', logo=temp_logo), + data={ + 'name': 'Test brand', + 'domain': 'bl.uk', + 'operation': 'branding-details' + }, + follow_redirects=True + ) + + assert response.status_code == 200 + + mock_letter_client.create_letter_branding.assert_called_once_with( + domain='bl.uk', filename='{}-test'.format(fake_uuid), name='Test brand' + ) + assert mock_template_preview.called + mock_persist_logo.assert_called_once_with( + temp_logo, + 'letters/static/images/letter-template/{}-test.svg'.format(fake_uuid) + ) + mock_upload_png.assert_called_once_with( + 'letters/static/images/letter-template/{}-test.png'.format(fake_uuid), + 'fake_png', + current_app.config['AWS_REGION'] + ) + mock_delete_temp_files.assert_called_once_with(user_id) + + +def test_create_letter_branding_shows_form_errors_on_name_and_domain_fields( + logged_in_platform_admin_client, + fake_uuid +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + temp_logo = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename='test.svg') + + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding', logo=temp_logo), + data={ + 'name': '', + 'domain': 'example.com', + 'operation': 'branding-details' + } + ) + + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + error_messages = page.find_all('span', class_='error-message') + + assert page.find('h1').text == 'Add letter branding' + assert len(error_messages) == 2 + assert error_messages[0].text.strip() == 'This field is required.' + assert error_messages[1].text.strip() == 'Not a known government domain (you might need to update domains.yml)' + + +@pytest.mark.parametrize('error_field', ['name', 'domain']) +def test_create_letter_branding_shows_database_errors_on_name_and_domain_fields( + mocker, + logged_in_platform_admin_client, + fake_uuid, + error_field +): + with logged_in_platform_admin_client.session_transaction() as session: + user_id = session["user_id"] + + mocker.patch('app.main.views.letter_branding.get_png_file_from_svg') + mocker.patch('app.main.views.letter_branding.letter_branding_client.create_letter_branding', side_effect=HTTPError( + response=Mock( + status_code=400, + json={ + 'result': 'error', + 'message': { + error_field: { + '{} already in use'.format(error_field) + } + } + } + ), + message={error_field: ['{} already in use'.format(error_field)]} + )) + + temp_logo = LETTER_TEMP_LOGO_LOCATION.format(user_id=user_id, unique_id=fake_uuid, filename='test.svg') + + response = logged_in_platform_admin_client.post( + url_for('.create_letter_branding', logo=temp_logo), + data={ + 'name': 'my brand', + 'domain': None, + 'operation': 'branding-details' + } + ) + + page = BeautifulSoup(response.data.decode('utf-8'), 'html.parser') + error_message = page.find('span', class_='error-message').text.strip() + + assert page.find('h1').text == 'Add letter branding' + assert error_message == '{} already in use'.format(error_field) + + +def test_get_png_file_from_svg(client, mocker, fake_uuid): + mocker.patch.dict( + 'flask.current_app.config', + {'TEMPLATE_PREVIEW_API_HOST': 'localhost', 'TEMPLATE_PREVIEW_API_KEY': 'abc'} + ) + tp_mock = mocker.patch('app.main.views.letter_branding.requests_get') + filename = LETTER_TEMP_LOGO_LOCATION.format(user_id=fake_uuid, unique_id=fake_uuid, filename='test.svg') + + get_png_file_from_svg(filename) + + tp_mock.assert_called_once_with( + 'localhost/temp-{}_{}-test.svg.png'.format(fake_uuid, fake_uuid), + headers={'Authorization': 'Token abc'} + ) diff --git a/tests/app/notify_client/test_letter_branding_client.py b/tests/app/notify_client/test_letter_branding_client.py index eceadf474..23f962971 100644 --- a/tests/app/notify_client/test_letter_branding_client.py +++ b/tests/app/notify_client/test_letter_branding_client.py @@ -7,3 +7,17 @@ def test_get_letter_branding(mocker): mock_get.assert_called_once_with( url='/dvla_organisations' ) + + +def test_create_letter_branding(mocker): + new_branding = {'filename': 'uuid-test', 'name': 'my letters', 'domain': 'example.com'} + + mock_post = mocker.patch('app.notify_client.letter_branding_client.LetterBrandingClient.post') + + LetterBrandingClient().create_letter_branding( + filename=new_branding['filename'], name=new_branding['name'], domain=new_branding['domain'] + ) + mock_post.assert_called_once_with( + url='/letter-branding', + data=new_branding + ) diff --git a/tests/app/s3_client/__init__.py b/tests/app/s3_client/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/app/s3_client/test_s3_csv_client.py b/tests/app/s3_client/test_s3_csv_client.py new file mode 100644 index 000000000..f4014065f --- /dev/null +++ b/tests/app/s3_client/test_s3_csv_client.py @@ -0,0 +1,21 @@ +from unittest.mock import Mock + +from app.s3_client.s3_csv_client import set_metadata_on_csv_upload + + +def test_sets_metadata(client, mocker): + mocked_s3_object = Mock() + mocked_get_s3_object = mocker.patch( + 'app.s3_client.s3_csv_client.get_csv_upload', + return_value=mocked_s3_object, + ) + + set_metadata_on_csv_upload('1234', '5678', foo='bar', baz=True) + + mocked_get_s3_object.assert_called_once_with('1234', '5678') + mocked_s3_object.copy_from.assert_called_once_with( + CopySource='test-notifications-csv-upload/service-1234-notify/5678.csv', + Metadata={'baz': 'True', 'foo': 'bar'}, + MetadataDirective='REPLACE', + ServerSideEncryption='AES256', + ) diff --git a/tests/app/s3_client/test_s3_logo_client.py b/tests/app/s3_client/test_s3_logo_client.py new file mode 100644 index 000000000..29cedd4a9 --- /dev/null +++ b/tests/app/s3_client/test_s3_logo_client.py @@ -0,0 +1,203 @@ +from collections import namedtuple +from unittest.mock import call + +import pytest + +from app.s3_client.s3_logo_client import ( + EMAIL_LOGO_LOCATION_STRUCTURE, + LETTER_TEMP_LOGO_LOCATION, + LETTER_TEMP_TAG, + TEMP_TAG, + delete_email_temp_file, + delete_email_temp_files_created_by, + delete_letter_temp_file, + delete_letter_temp_files_created_by, + letter_filename_for_db, + permanent_email_logo_name, + persist_logo, + upload_email_logo, + upload_letter_png_logo, + upload_letter_temp_logo, +) + +bucket = 'test_bucket' +data = {'data': 'some_data'} +filename = 'test.png' +svg_filename = 'test.svg' +upload_id = 'test_uuid' +region = 'eu-west1' + + +@pytest.fixture +def upload_filename(fake_uuid): + return EMAIL_LOGO_LOCATION_STRUCTURE.format( + temp=TEMP_TAG.format(user_id=fake_uuid), unique_id=upload_id, filename=filename) + + +@pytest.fixture +def letter_upload_filename(fake_uuid): + return LETTER_TEMP_LOGO_LOCATION.format( + user_id=fake_uuid, + unique_id=upload_id, + filename=svg_filename + ) + + +def test_upload_email_logo_calls_correct_args(client, mocker, fake_uuid, upload_filename): + mocker.patch('uuid.uuid4', return_value=upload_id) + mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) + mocked_s3_upload = mocker.patch('app.s3_client.s3_logo_client.utils_s3upload') + + upload_email_logo(filename=filename, user_id=fake_uuid, filedata=data, region=region) + + mocked_s3_upload.assert_called_once_with( + filedata=data, + region=region, + file_location=upload_filename, + bucket_name=bucket, + content_type='image/png' + ) + + +def test_upload_letter_temp_logo_calls_correct_args(mocker, fake_uuid, letter_upload_filename): + mocker.patch('uuid.uuid4', return_value=upload_id) + mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) + mocked_s3_upload = mocker.patch('app.s3_client.s3_logo_client.utils_s3upload') + + new_filename = upload_letter_temp_logo(filename=svg_filename, user_id=fake_uuid, filedata=data, region=region) + + mocked_s3_upload.assert_called_once_with( + filedata=data, + region=region, + bucket_name=bucket, + file_location=letter_upload_filename, + content_type='image/svg+xml' + ) + assert new_filename == 'letters/static/images/letter-template/temp-{}_test_uuid-test.svg'.format(fake_uuid) + + +def test_upload_letter_png_logo_calls_correct_args(mocker): + mocked_s3_upload = mocker.patch('app.s3_client.s3_logo_client.utils_s3upload') + mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) + + upload_letter_png_logo(filename, data, region) + + mocked_s3_upload.assert_called_once_with( + filedata=data, + region=region, + bucket_name=bucket, + file_location=filename, + content_type='image/png' + ) + + +def test_persist_logo(client, mocker, fake_uuid, upload_filename): + mocker.patch.dict('flask.current_app.config', {'LOGO_UPLOAD_BUCKET_NAME': bucket}) + mocked_get_s3_object = mocker.patch('app.s3_client.s3_logo_client.get_s3_object') + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + new_filename = permanent_email_logo_name(upload_filename, fake_uuid) + + persist_logo(upload_filename, new_filename) + + mocked_get_s3_object.assert_called_once_with(bucket, new_filename) + mocked_delete_s3_object.assert_called_once_with(upload_filename) + + +def test_persist_logo_returns_if_not_temp(client, mocker, fake_uuid): + filename = 'logo.png' + persist_logo(filename, filename) + + mocked_get_s3_object = mocker.patch('app.s3_client.s3_logo_client.get_s3_object') + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + mocked_get_s3_object.assert_not_called() + mocked_delete_s3_object.assert_not_called() + + +def test_permanent_email_logo_name_removes_TEMP_TAG_from_filename(upload_filename, fake_uuid): + new_name = permanent_email_logo_name(upload_filename, fake_uuid) + + assert new_name == 'test_uuid-test.png' + + +def test_permanent_email_logo_name_does_not_change_filenames_with_no_TEMP_TAG(): + filename = 'logo.png' + new_name = permanent_email_logo_name(filename, filename) + + assert new_name == filename + + +def test_letter_filename_for_db_when_file_has_a_temp_tag(fake_uuid): + temp_filename = LETTER_TEMP_LOGO_LOCATION.format(user_id=fake_uuid, unique_id=upload_id, filename=svg_filename) + assert letter_filename_for_db(temp_filename, fake_uuid) == 'test_uuid-test' + + +def test_letter_filename_for_db_when_file_does_not_have_a_temp_tag(fake_uuid): + filename = 'letters/static/images/letter-template/{}-test.svg'.format(fake_uuid) + assert letter_filename_for_db(filename, fake_uuid) == '{}-test'.format(fake_uuid) + + +def test_delete_email_temp_files_created_by_user(client, mocker, fake_uuid): + obj = namedtuple("obj", ["key"]) + objs = [obj(key='test1'), obj(key='test2')] + + mocker.patch('app.s3_client.s3_logo_client.get_s3_objects_filter_by_prefix', return_value=objs) + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + delete_email_temp_files_created_by(fake_uuid) + + for index, arg in enumerate(mocked_delete_s3_object.call_args_list): + assert arg == call(objs[index].key) + + +def test_delete_letter_temp_files_created_by_user(mocker, fake_uuid): + obj = namedtuple("obj", ["key"]) + objs = [obj(key='test1'), obj(key='test2')] + + mocker.patch('app.s3_client.s3_logo_client.get_s3_objects_filter_by_prefix', return_value=objs) + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + delete_letter_temp_files_created_by(fake_uuid) + + for index, arg in enumerate(mocked_delete_s3_object.call_args_list): + assert arg == call(objs[index].key) + + +def test_delete_single_email_temp_file(client, mocker, upload_filename): + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + delete_email_temp_file(upload_filename) + + mocked_delete_s3_object.assert_called_with(upload_filename) + + +def test_does_not_delete_non_temp_email_file(client, mocker): + filename = 'logo.png' + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + with pytest.raises(ValueError) as error: + delete_email_temp_file(filename) + + mocked_delete_s3_object.assert_not_called + assert str(error.value) == 'Not a temp file: {}'.format(filename) + + +def test_delete_single_temp_letter_file(mocker, fake_uuid, upload_filename): + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + upload_filename = LETTER_TEMP_TAG.format(user_id=fake_uuid) + svg_filename + + delete_letter_temp_file(upload_filename) + + mocked_delete_s3_object.assert_called_with(upload_filename) + + +def test_does_not_delete_non_temp_letter_file(mocker, fake_uuid): + mocked_delete_s3_object = mocker.patch('app.s3_client.s3_logo_client.delete_s3_object') + + with pytest.raises(ValueError) as error: + delete_letter_temp_file(svg_filename) + + mocked_delete_s3_object.assert_not_called + assert str(error.value) == 'Not a temp file: {}'.format(svg_filename) diff --git a/tests/app/test_utils.py b/tests/app/test_utils.py index d65cafbb5..7bdfd4ebf 100644 --- a/tests/app/test_utils.py +++ b/tests/app/test_utils.py @@ -210,7 +210,7 @@ def test_generate_notifications_csv_returns_correct_csv_file( expected_1st_row, ): mocker.patch( - 'app.main.s3_client.s3download', + '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') @@ -236,7 +236,7 @@ def test_generate_notifications_csv_calls_twice_if_next_link( ): mocker.patch( - 'app.main.s3_client.s3download', + 'app.s3_client.s3_csv_client.s3download', return_value=""" phone_number 07700900000