Merge pull request #3220 from alphagov/fix-broadcast-length-validation

Normalise broadcast content before validating length

app/broadcast_message/broadcast_message_schema.py

@@ -15,7 +15,7 @@ create_broadcast_message_schema = {
         'finishes_at': {'type': 'string', 'format': 'datetime'},
         'areas': {"type": "array", "items": {"type": "string"}},
         'simple_polygons': {"type": "array", "items": {"type": "array"}},
-        'content': {'type': 'string', 'minLength': 1, 'maxLength': 1395},
+        'content': {'type': 'string', 'minLength': 1},
         'reference': {'type': 'string', 'minLength': 1, 'maxLength': 255},
     },
     'required': ['service_id', 'created_by'],

app/broadcast_message/rest.py

@@ -2,6 +2,7 @@ from datetime import datetime
 
 import iso8601
 from flask import Blueprint, current_app, jsonify, request
+from notifications_utils.template import BroadcastMessageTemplate
 
 from app.broadcast_message.broadcast_message_schema import (
     create_broadcast_message_schema,
@@ -116,6 +117,19 @@ def create_broadcast_message(service_id):
         reference = None
     else:
         template, content, reference = None, data['content'], data['reference']
+        temporary_template = BroadcastMessageTemplate.from_content(content)
+        if temporary_template.content_too_long:
+            raise InvalidRequest(
+                (
+                    f'Content must be '
+                    f'{temporary_template.max_content_count:,.0f} '
+                    f'characters or fewer'
+                ) + (
+                    ' (because it could not be GSM7 encoded)'
+                    if temporary_template.non_gsm_characters else ''
+                ),
+                status_code=400,
+            )
 
     broadcast_message = BroadcastMessage(
         service_id=service.id,

app/template/rest.py

@@ -5,7 +5,10 @@ import botocore
 from flask import Blueprint, current_app, jsonify, request
 from notifications_utils import SMS_CHAR_COUNT_LIMIT
 from notifications_utils.pdf import extract_page_from_pdf
-from notifications_utils.template import SMSMessageTemplate
+from notifications_utils.template import (
+    BroadcastMessageTemplate,
+    SMSMessageTemplate,
+)
 from PyPDF2.utils import PdfReadError
 from requests import post as requests_post
 from sqlalchemy.orm.exc import NoResultFound
@@ -28,7 +31,13 @@ from app.dao.templates_dao import (
 )
 from app.errors import InvalidRequest, register_errors
 from app.letters.utils import get_letter_pdf_and_metadata
-from app.models import LETTER_TYPE, SECOND_CLASS, SMS_TYPE, Template
+from app.models import (
+    BROADCAST_TYPE,
+    LETTER_TYPE,
+    SECOND_CLASS,
+    SMS_TYPE,
+    Template,
+)
 from app.notifications.validators import check_reply_to, service_has_permission
 from app.schema_validation import validate
 from app.schemas import (
@@ -48,10 +57,13 @@ register_errors(template_blueprint)
 
 
 def _content_count_greater_than_limit(content, template_type):
-    if template_type != SMS_TYPE:
-        return False
-    template = SMSMessageTemplate({'content': content, 'template_type': template_type})
-    return template.is_message_too_long()
+    if template_type == SMS_TYPE:
+        template = SMSMessageTemplate({'content': content, 'template_type': template_type})
+        return template.is_message_too_long()
+    if template_type == BROADCAST_TYPE:
+        template = BroadcastMessageTemplate({'content': content, 'template_type': template_type})
+        return template.is_message_too_long()
+    return False
 
 
 def validate_parent_folder(template_json):

tests/app/broadcast_message/test_rest.py

@@ -192,6 +192,40 @@ def test_create_broadcast_message_400s_if_json_schema_fails_validation(
     assert response['errors'] == expected_errors
 
 
+@pytest.mark.parametrize('content, expected_status, expected_errors', (
+    ('a', 201, None),
+    ('a' * 1_395, 201, None),
+    ('a\r\n' * 697, 201, None),  # 1,394 chars – new lines normalised to \n
+    ('a' * 1_396, 400, (
+        'Content must be 1,395 characters or fewer'
+    )),
+    ('ŵ' * 615, 201, None),
+    ('ŵ' * 616, 400, (
+        'Content must be 615 characters or fewer '
+        '(because it could not be GSM7 encoded)'
+    )),
+))
+def test_create_broadcast_message_400s_if_content_too_long(
+    admin_request,
+    sample_broadcast_service,
+    content,
+    expected_status,
+    expected_errors,
+):
+    response = admin_request.post(
+        'broadcast_message.create_broadcast_message',
+        service_id=sample_broadcast_service.id,
+        _data={
+            'service_id': str(sample_broadcast_service.id),
+            'created_by': str(sample_broadcast_service.created_by_id),
+            'reference': 'abc123',
+            'content': content,
+        },
+        _expected_status=expected_status,
+    )
+    assert response.get('message') == expected_errors
+
+
 @freeze_time('2020-01-01')
 def test_create_broadcast_message_can_be_created_from_content(admin_request, sample_broadcast_service):
     response = admin_request.post(

tests/app/template/test_rest.py

@@ -731,14 +731,24 @@ def test_should_return_404_if_no_templates_for_service_with_id(client, sample_se
     assert json_resp['message'] == 'No result found'
 
 
-def test_create_400_for_over_limit_content(client, notify_api, sample_user, sample_service, fake_uuid):
+@pytest.mark.parametrize('template_type', (
+    SMS_TYPE, BROADCAST_TYPE,
+))
+def test_create_400_for_over_limit_content(
+    client,
+    notify_api,
+    sample_user,
+    fake_uuid,
+    template_type,
+):
+    sample_service = create_service(service_permissions=[template_type])
     content = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(SMS_CHAR_COUNT_LIMIT + 1))
     data = {
         'name': 'too big template',
-        'template_type': SMS_TYPE,
+        'template_type': template_type,
         'content': content,
         'service': str(sample_service.id),
-        'created_by': str(sample_user.id)
+        'created_by': str(sample_service.created_by.id)
     }
     data = json.dumps(data)
     auth_header = create_authorization_header()