Add a canary app for testing

using lots of CPU by tracing the application to gather some data

Alternatively we could take a stab in the dark, which is what this
commit is doing.

I have the hypothesis that we are not re-using TCP connections using
HTTP keepalive

Refer to https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Keep-Alive

This means we are renegotiating the TLS connection every time we want to
send an SMS.  When we are sending lots of SMS messages then this will do
a lot of crypto handshaking which is expensive (in terms of CPU)

ie send_sms calls request() which creates a new tcp connection and a
new TLS handshake

When you use request.Session() to create a session it uses urllib3's
connection pooling, I've arbitrarily chosen 32 connections per pool
(with a default number of pools = 10)

ie init_app creates a session which has underlying connection pools
send_sms claims a connection from the pool and uses it to create or
re-use an existing TLS connection

Sessions are usually not great because they share data like cookies, but
when calling an API this is fine, or at least it is probably worth
canarying

Another way of re-using connections is by running HAProxy or similar as
a side-car proxy, which proxies to the API. send_sms would make a
local TCP connection to HAProxy which proxies to the MMG or Firetext
API via TLS, adding the Connection: keep-alive header

This command can be used to see how many TLS handshakes your computer is
doing, with some false positives:

tcpdump -n "tcp port 443 and (tcp[((tcp[12] & 0xf0) >> 2)] = 0x16)"

(alternatively we could just instrument the code /shrug)

Signed-off-by: toby lorne <toby@toby.codes>

app/broadcast_message/broadcast_message_schema.py

@@ -15,8 +15,20 @@ 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},
+        'reference': {'type': 'string', 'minLength': 1, 'maxLength': 255},
     },
-    'required': ['template_id', 'service_id', 'created_by'],
+    'required': ['service_id', 'created_by'],
+    'allOf': [
+        {'oneOf': [
+            {'required': ['template_id']},
+            {'required': ['content']},
+        ]},
+        {'oneOf': [
+            {'required': ['template_id']},
+            {'required': ['reference']},
+        ]},
+    ],
     'additionalProperties': False
 }
 

app/broadcast_message/rest.py

@@ -49,12 +49,8 @@ def _update_broadcast_message(broadcast_message, new_status, updating_user):
         )
 
     if new_status == BroadcastStatusType.BROADCASTING:
-        # TODO: Remove this platform admin shortcut when the feature goes live
-        if updating_user == broadcast_message.created_by and not (
-            # platform admins and trial mode services can approve their own broadcasts
-            updating_user.platform_admin or
-            broadcast_message.service.restricted
-        ):
+        # training mode services can approve their own broadcasts
+        if updating_user == broadcast_message.created_by and not broadcast_message.service.restricted:
             raise InvalidRequest(
                 f'User {updating_user.id} cannot approve their own broadcast_message {broadcast_message.id}',
                 status_code=400
@@ -99,22 +95,32 @@ def create_broadcast_message(service_id):
     validate(data, create_broadcast_message_schema)
     service = dao_fetch_service_by_id(data['service_id'])
     user = get_user_by_id(data['created_by'])
-    template = dao_get_template_by_id_and_service_id(data['template_id'], data['service_id'])
-
     personalisation = data.get('personalisation', {})
+    template_id = data.get('template_id')
+
+    if template_id:
+        template = dao_get_template_by_id_and_service_id(
+            template_id, data['service_id']
+        )
+        content = template._as_utils_template_with_personalisation(
+            personalisation
+        ).content_with_placeholders_filled_in
+        reference = None
+    else:
+        template, content, reference = None, data['content'], data['reference']
+
     broadcast_message = BroadcastMessage(
         service_id=service.id,
-        template_id=template.id,
-        template_version=template.version,
+        template_id=template_id,
+        template_version=template.version if template else None,
         personalisation=personalisation,
         areas={"areas": data.get("areas", []), "simple_polygons": data.get("simple_polygons", [])},
         status=BroadcastStatusType.DRAFT,
         starts_at=_parse_nullable_datetime(data.get('starts_at')),
         finishes_at=_parse_nullable_datetime(data.get('finishes_at')),
         created_by_id=user.id,
-        content=template._as_utils_template_with_personalisation(
-            personalisation
-        ).content_with_placeholders_filled_in,
+        content=content,
+        reference=reference,
     )
 
     dao_save_object(broadcast_message)

app/celery/tasks.py

@@ -42,7 +42,7 @@ from app.dao.returned_letters_dao import insert_or_update_returned_letters
 from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_id
 from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service
 from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
-from app.dao.services_dao import dao_fetch_service_by_id, fetch_todays_total_message_count
+from app.dao.services_dao import fetch_todays_total_message_count
 from app.dao.templates_dao import dao_get_template_by_id
 from app.exceptions import DVLAException, NotificationTechnicalFailureException
 from app.models import (
@@ -65,6 +65,7 @@ from app.models import (
 )
 from app.notifications.process_notifications import persist_notification
 from app.service.utils import service_allowed_to_send_to
+from app.serialised_models import SerialisedService, SerialisedTemplate
 from app.utils import DATETIME_FORMAT
 
 
@@ -190,13 +191,17 @@ def save_sms(self,
              encrypted_notification,
              sender_id=None):
     notification = encryption.decrypt(encrypted_notification)
-    service = dao_fetch_service_by_id(service_id)
-    template = dao_get_template_by_id(notification['template'], version=notification['template_version'])
+    service = SerialisedService.from_id(service_id)
+    template = SerialisedTemplate.from_id_and_service_id(
+        notification['template'],
+        service_id=service.id,
+        version=notification['template_version'],
+    )
 
     if sender_id:
         reply_to_text = dao_get_service_sms_senders_by_id(service_id, sender_id).sms_sender
     else:
-        reply_to_text = template.get_reply_to_text()
+        reply_to_text = template.reply_to_text
 
     if not service_allowed_to_send_to(notification['to'], service, KEY_TYPE_NORMAL):
         current_app.logger.debug(
@@ -246,13 +251,17 @@ def save_email(self,
                sender_id=None):
     notification = encryption.decrypt(encrypted_notification)
 
-    service = dao_fetch_service_by_id(service_id)
-    template = dao_get_template_by_id(notification['template'], version=notification['template_version'])
+    service = SerialisedService.from_id(service_id)
+    template = SerialisedTemplate.from_id_and_service_id(
+        notification['template'],
+        service_id=service.id,
+        version=notification['template_version'],
+    )
 
     if sender_id:
         reply_to_text = dao_get_reply_to_by_id(service_id, sender_id).email_address
     else:
-        reply_to_text = template.get_reply_to_text()
+        reply_to_text = template.reply_to_text
 
     if not service_allowed_to_send_to(notification['to'], service, KEY_TYPE_NORMAL):
         current_app.logger.info("Email {} failed as restricted service".format(notification_id))
@@ -300,7 +309,7 @@ def save_api_sms(self, encrypted_notification):
 
 def save_api_email_or_sms(self, encrypted_notification):
     notification = encryption.decrypt(encrypted_notification)
-    service = dao_fetch_service_by_id(notification['service_id'])
+    service = SerialisedService.from_id(notification['service_id'])
     q = QueueNames.SEND_EMAIL if notification['notification_type'] == EMAIL_TYPE else QueueNames.SEND_SMS
     provider_task = provider_tasks.deliver_email if notification['notification_type'] == EMAIL_TYPE \
         else provider_tasks.deliver_sms
@@ -356,8 +365,12 @@ def save_letter(
         Columns(notification['personalisation'])
     )
 
-    service = dao_fetch_service_by_id(service_id)
-    template = dao_get_template_by_id(notification['template'], version=notification['template_version'])
+    service = SerialisedService.from_id(service_id)
+    template = SerialisedTemplate.from_id_and_service_id(
+        notification['template'],
+        service_id=service.id,
+        version=notification['template_version'],
+    )
 
     try:
         # if we don't want to actually send the letter, then start it off in SENDING so we don't pick it up
@@ -378,7 +391,7 @@ def save_letter(
             job_row_number=notification['row_number'],
             notification_id=notification_id,
             reference=create_random_identifier(),
-            reply_to_text=template.get_reply_to_text(),
+            reply_to_text=template.reply_to_text,
             status=status
         )
 

app/clients/cbc_proxy.py

@@ -46,6 +46,7 @@ class CBCProxyClient:
             'canary': CBCProxyCanary,
             BroadcastProvider.EE: CBCProxyEE,
             BroadcastProvider.THREE: CBCProxyThree,
+            BroadcastProvider.O2: CBCProxyO2,
             BroadcastProvider.VODAFONE: CBCProxyVodafone,
         }
         return proxy_classes[provider](self._lambda_client)
@@ -131,13 +132,13 @@ class CBCProxyClientBase(ABC):
 
         if result['StatusCode'] > 299:
             current_app.logger.info(
-                f"Error calling lambda {self.lambda_name} with status code { result['StatusCode']}, {result.get('Payload')}"
+                f"Error calling lambda {lambda_name} with status code { result['StatusCode']}, {result.get('Payload')}"
             )
             success = False
 
         elif 'FunctionError' in result:
             current_app.logger.info(
-                f"Error calling lambda {self.lambda_name} with function error { result['Payload'] }"
+                f"Error calling lambda {lambda_name} with function error { result['Payload'] }"
             )
             success = False
 
@@ -291,6 +292,65 @@ class CBCProxyThree(CBCProxyClientBase):
         }
         self._invoke_lambda_with_failover(payload=payload)
 
+class CBCProxyO2(CBCProxyClientBase):
+    lambda_name = 'o2-1-proxy'
+    failover_lambda_name = 'o2-2-proxy'
+
+    LANGUAGE_ENGLISH = 'en-GB'
+    LANGUAGE_WELSH = 'cy-GB'
+
+    def send_link_test(
+        self,
+        identifier,
+        sequential_number=None,
+    ):
+        """
+        link test - open up a connection to a specific provider, and send them an xml payload with a <msgType> of
+        test.
+        """
+        payload = {
+            'message_type': 'test',
+            'identifier': identifier,
+            'message_format': 'cap'
+        }
+
+        self._invoke_lambda_with_failover(payload=payload)
+
+    def create_and_send_broadcast(
+        self, identifier, headline, description, areas, sent, expires, message_number=None
+    ):
+        payload = {
+            'message_type': 'alert',
+            'identifier': identifier,
+            'message_format': 'cap',
+            'headline': headline,
+            'description': description,
+            'areas': areas,
+            'sent': sent,
+            'expires': expires,
+            'language': self.infer_language_from(description),
+        }
+        self._invoke_lambda_with_failover(payload=payload)
+
+    def cancel_broadcast(
+        self,
+        identifier, previous_provider_messages,
+        sent, message_number=None
+    ):
+        payload = {
+            'message_type': 'cancel',
+            'identifier': identifier,
+            'message_format': 'cap',
+            "references": [
+                {
+                    "message_id": str(message.id),
+                    "sent": message.created_at.strftime(DATETIME_FORMAT)
+                } for message in previous_provider_messages
+            ],
+            'sent': sent,
+        }
+        self._invoke_lambda_with_failover(payload=payload)
+
 
 class CBCProxyVodafone(CBCProxyClientBase):
     lambda_name = 'vodafone-1-proxy'

app/clients/email/aws_ses.py

@@ -62,6 +62,7 @@ class AwsSesClient(EmailClient):
         # before-call, after-call, after-call-error, request-created, response-received
         self._client.meta.events.register('request-created.ses.SendEmail', self.ses_request_created_hook)
         self._client.meta.events.register('response-received.ses.SendEmail', self.ses_response_received_hook)
+        self._client.meta.events.register('before-call.ses', self.ses_inject_connection_header)
 
     def ses_request_created_hook(self, **kwargs):
         # request created may be called multiple times if the request auto-retries. We want to count all these as the
@@ -77,6 +78,11 @@ class AwsSesClient(EmailClient):
     def get_name(self):
         return self.name
 
+    def ses_inject_connection_header(self, params, **kwargs):
+        # keep underlying TLS connection open, so we do not spend lots of CPU
+        # and network time renegotiating TLS
+        params['headers']['Connection'] = 'Keep-Alive'
+
     def send_email(self,
                    source,
                    to_addresses,
@@ -148,3 +154,4 @@ def punycode_encode_email(email_address):
     # only the hostname should ever be punycode encoded.
     local, hostname = email_address.split('@')
     return '{}@{}'.format(local, hostname.encode('idna').decode('utf-8'))
+

app/clients/sms/firetext.py

@@ -2,7 +2,8 @@ import json
 import logging
 
 from time import monotonic
-from requests import request, RequestException
+from requests import request, RequestException, Session
+from requests.adapters import HTTPAdapter
 
 from app.clients.sms import (SmsClient, SmsClientResponseException)
 
@@ -69,6 +70,9 @@ class FiretextClient(SmsClient):
         self.name = 'firetext'
         self.url = current_app.config.get('FIRETEXT_URL')
         self.statsd_client = statsd_client
+        # this uses urllib3 under the hood to create a connection pool
+        self.session = Session()
+        self.session.mount('https://', HTTPAdapter(pool_maxsize=32))
 
     def get_name(self):
         return self.name
@@ -103,8 +107,7 @@ class FiretextClient(SmsClient):
         response = None
         start_time = monotonic()
         try:
-            response = request(
-                "POST",
+            response = self.session.post(
                 self.url,
                 data=data,
                 timeout=60

app/clients/sms/mmg.py

@@ -1,6 +1,8 @@
 import json
 from time import monotonic
-from requests import (request, RequestException)
+from requests import (request, RequestException, Session)
+from requests.adapters import HTTPAdapter
+
 from app.clients.sms import (SmsClient, SmsClientResponseException)
 
 mmg_response_map = {
@@ -75,6 +77,9 @@ class MMGClient(SmsClient):
         self.name = 'mmg'
         self.statsd_client = statsd_client
         self.mmg_url = current_app.config.get('MMG_URL')
+        # this uses urllib3 under the hood to create a connection pool
+        self.session = Session()
+        self.session.mount('https://', HTTPAdapter(pool_maxsize=32))
 
     def record_outcome(self, success, response):
         status_code = response.status_code if response else 503
@@ -108,8 +113,8 @@ class MMGClient(SmsClient):
         response = None
         start_time = monotonic()
         try:
-            response = request(
-                "POST",
+
+            response = self.session.post(
                 self.mmg_url,
                 data=json.dumps(data),
                 headers={

app/config.py

@@ -380,7 +380,7 @@ class Config(object):
 
     CBC_PROXY_ENABLED = bool(CBC_PROXY_AWS_ACCESS_KEY_ID)
 
-    ENABLED_CBCS = {BroadcastProvider.EE, BroadcastProvider.THREE, BroadcastProvider.VODAFONE}
+    ENABLED_CBCS = {BroadcastProvider.EE, BroadcastProvider.THREE, BroadcastProvider.O2, BroadcastProvider.VODAFONE}
 
 
 ######################

app/schemas.py

@@ -115,13 +115,13 @@ class UserSchema(BaseSchema):
     class Meta(BaseSchema.Meta):
         model = models.User
         exclude = (
-            "updated_at",
-            "email_access_validated_at",
-            "created_at",
-            "user_to_service",
-            "user_to_organisation",
             "_password",
-            "verify_codes"
+            "created_at",
+            "email_access_validated_at",
+            "updated_at",
+            "user_to_organisation",
+            "user_to_service",
+            "verify_codes",
         )
         strict = True
 
@@ -152,9 +152,18 @@ class UserUpdateAttributeSchema(BaseSchema):
     class Meta(BaseSchema.Meta):
         model = models.User
         exclude = (
-            'id', 'updated_at', 'created_at', 'user_to_service',
-            '_password', 'verify_codes', 'logged_in_at', 'password_changed_at',
-            'failed_login_count', 'state', 'platform_admin')
+            '_password',
+            'created_at',
+            'failed_login_count',
+            'id',
+            'logged_in_at',
+            'password_changed_at',
+            'platform_admin',
+            'state',
+            'updated_at',
+            'user_to_service',
+            'verify_codes',
+        )
         strict = True
 
     @validates('name')
@@ -243,33 +252,33 @@ class ServiceSchema(BaseSchema, UUIDsAsStringsMixin):
     class Meta(BaseSchema.Meta):
         model = models.Service
         exclude = (
-            'updated_at',
-            'created_at',
-            'api_keys',
-            'templates',
-            'jobs',
-            'old_id',
-            'template_statistics',
-            'service_provider_stats',
-            'service_notification_stats',
-            'service_sms_senders',
-            'reply_to_email_addresses',
-            'letter_contacts',
-            'complaints',
-            'data_retention',
             'all_template_folders',
             'annual_billing',
+            'api_keys',
+            'broadcast_messages',
+            'complaints',
             'contact_list',
+            'created_at',
             'crown',
+            'data_retention',
+            'guest_list',
             'inbound_number',
             'inbound_sms',
+            'jobs',
+            'letter_contacts',
             'letter_logo_filename',
+            'old_id',
+            'reply_to_email_addresses',
             'returned_letters',
+            'service_broadcast_provider_restriction',
+            'service_notification_stats',
+            'service_provider_stats',
+            'service_sms_senders',
+            'template_statistics',
+            'templates',
+            'updated_at',
             'users',
             'version',
-            'guest_list',
-            'broadcast_messages',
-            'service_broadcast_provider_restriction',
         )
         strict = True
 
@@ -303,40 +312,37 @@ class DetailedServiceSchema(BaseSchema):
     class Meta(BaseSchema.Meta):
         model = models.Service
         exclude = (
-            'api_keys',
-            'templates',
-            'users',
-            'created_by',
-            'jobs',
-            'template_statistics',
-            'service_provider_stats',
-            'service_notification_stats',
-            'email_branding',
-            'service_sms_senders',
-            'monthly_billing',
-            'reply_to_email_addresses',
-            'letter_contact_block',
-            'message_limit',
-            'email_from',
-            'inbound_api',
-            'guest_list',
-            'reply_to_email_address',
-            'sms_sender',
-            'permissions',
-            'inbound_number',
-            'inbound_sms',
             'all_template_folders',
             'annual_billing',
+            'api_keys',
+            'broadcast_messages',
             'contact_list',
             'created_by',
             'crown',
+            'email_branding',
+            'email_from',
+            'guest_list',
+            'inbound_api',
+            'inbound_number',
+            'inbound_sms',
+            'jobs',
+            'letter_contact_block',
             'letter_logo_filename',
+            'message_limit',
+            'monthly_billing',
+            'permissions',
             'rate_limit',
+            'reply_to_email_address',
+            'reply_to_email_addresses',
             'returned_letters',
+            'service_notification_stats',
+            'service_provider_stats',
+            'service_sms_senders',
+            'sms_sender',
+            'template_statistics',
+            'templates',
             'users',
             'version',
-            'guest_list',
-            'broadcast_messages',
         )
 
 
@@ -386,6 +392,7 @@ class TemplateSchemaNoDetail(TemplateSchema):
     class Meta(TemplateSchema.Meta):
         exclude = TemplateSchema.Meta.exclude + (
             'archived',
+            'broadcast_data',
             'created_at',
             'created_by',
             'created_by_id',
@@ -401,7 +408,6 @@ class TemplateSchemaNoDetail(TemplateSchema):
             'template_redacted',
             'updated_at',
             'version',
-            'broadcast_data',
         )
 
     @pre_dump
@@ -470,9 +476,10 @@ class JobSchema(BaseSchema):
         model = models.Job
         exclude = (
             'notifications',
-            'notifications_sent',
             'notifications_delivered',
-            'notifications_failed')
+            'notifications_failed',
+            'notifications_sent',
+        )
         strict = True
 
 
@@ -565,12 +572,25 @@ class NotificationWithPersonalisationSchema(NotificationWithTemplateSchema):
         # 'body', 'subject' [for emails], and 'content_char_count'
         fields = (
             # db rows
-            'id', 'to', 'job_row_number', 'template_version', 'billable_units', 'notification_type', 'created_at',
-            'sent_at', 'sent_by', 'updated_at', 'status', 'reference',
+            'billable_units',
+            'created_at',
+            'id',
+            'job_row_number',
+            'notification_type',
+            'reference',
+            'sent_at',
+            'sent_by',
+            'status',
+            'template_version',
+            'to',
+            'updated_at',
             # computed fields
             'personalisation',
             # relationships
-            'service', 'job', 'api_key', 'template_history'
+            'api_key',
+            'job',
+            'service',
+            'template_history',
         )
 
     @pre_dump

app/serialised_models.py

@@ -51,18 +51,19 @@ class SerialisedTemplate(SerialisedModel):
 
     @classmethod
     @memory_cache
-    def from_id_and_service_id(cls, template_id, service_id):
-        return cls(cls.get_dict(template_id, service_id)['data'])
+    def from_id_and_service_id(cls, template_id, service_id, version=None):
+        return cls(cls.get_dict(template_id, service_id, version)['data'])
 
     @staticmethod
-    @redis_cache.set('service-{service_id}-template-{template_id}-version-None')
-    def get_dict(template_id, service_id):
+    @redis_cache.set('service-{service_id}-template-{template_id}-version-{version}')
+    def get_dict(template_id, service_id, version):
         from app.dao import templates_dao
         from app.schemas import template_schema
 
         fetched_template = templates_dao.dao_get_template_by_id_and_service_id(
             template_id=template_id,
-            service_id=service_id
+            service_id=service_id,
+            version=version,
         )
 
         template_dict = template_schema.dump(fetched_template).data

manifest.yml.j2

@@ -53,6 +53,11 @@
   'notify-delivery-worker-jobs': {},
   'notify-delivery-worker-research': {},
   'notify-delivery-worker-sender': {'disk_quota': '2G', 'memory': '4G'},
+  'notify-delivery-worker-sender-canary': {'disk_quota': '2G', 'memory': '4G',  'instances': {
+      'preview': 0,
+      'staging': 1,
+      'production': 0
+    },},
   'notify-delivery-worker-periodic': {},
   'notify-delivery-worker-reporting': {
     'additional_env_vars': {

scripts/paas_app_wrapper.sh

@@ -24,6 +24,10 @@ case $NOTIFY_APP_NAME in
     exec scripts/run_multi_worker_app_paas.sh celery multi start 3 -c 10 -A run_celery.notify_celery --loglevel=INFO \
     --logfile=/dev/null --pidfile=/tmp/celery%N.pid -Q send-sms-tasks,send-email-tasks
     ;;
+    delivery-worker-sender-canary)
+    exec scripts/run_multi_worker_app_paas.sh celery multi start 3 -c 10 -A run_celery.notify_celery --loglevel=INFO \
+    --logfile=/dev/null --pidfile=/tmp/celery%N.pid -Q send-sms-tasks,send-email-tasks
+    ;;
   delivery-worker-periodic)
     exec scripts/run_app_paas.sh celery -A run_celery.notify_celery worker --loglevel=INFO --concurrency=2 \
     -Q periodic-tasks 2> /dev/null

tests/app/broadcast_message/test_rest.py

@@ -143,9 +143,9 @@ def test_create_broadcast_message(admin_request, sample_broadcast_service):
     (
         {},
         [
-            {'error': 'ValidationError', 'message': 'template_id is a required property'},
             {'error': 'ValidationError', 'message': 'service_id is a required property'},
-            {'error': 'ValidationError', 'message': 'created_by is a required property'}
+            {'error': 'ValidationError', 'message': 'created_by is a required property'},
+            {'error': 'ValidationError', 'message': '{} is not valid under any of the given schemas'},
         ]
     ),
     (
@@ -178,12 +178,32 @@ def test_create_broadcast_message_400s_if_json_schema_fails_validation(
 
 
 @freeze_time('2020-01-01')
-def test_create_broadcast_message_400s_if_content_provided_but_no_template(admin_request, sample_broadcast_service):
-    # we don't currently support this, but might in the future
+def test_create_broadcast_message_can_be_created_from_content(admin_request, sample_broadcast_service):
     response = admin_request.post(
         'broadcast_message.create_broadcast_message',
         _data={
-            'template_id': None,
+            'content': 'Some tailor made broadcast content',
+            'reference': 'abc123',
+            'service_id': str(sample_broadcast_service.id),
+            'created_by': str(sample_broadcast_service.created_by_id),
+        },
+        service_id=sample_broadcast_service.id,
+        _expected_status=201
+    )
+    assert response['content'] == 'Some tailor made broadcast content'
+    assert response['reference'] == 'abc123'
+    assert response['template_id'] is None
+
+
+def test_create_broadcast_message_400s_if_content_and_template_provided(
+    admin_request,
+    sample_broadcast_service,
+):
+    template = create_template(sample_broadcast_service, BROADCAST_TYPE)
+    response = admin_request.post(
+        'broadcast_message.create_broadcast_message',
+        _data={
+            'template_id': str(template.id),
             'content': 'Some tailor made broadcast content',
             'service_id': str(sample_broadcast_service.id),
             'created_by': str(sample_broadcast_service.created_by_id),
@@ -191,10 +211,93 @@ def test_create_broadcast_message_400s_if_content_provided_but_no_template(admin
         service_id=sample_broadcast_service.id,
         _expected_status=400
     )
-    assert response['errors'] ==[
-        {'error': 'ValidationError', 'message': 'template_id is not a valid UUID'},
-        {'error': 'ValidationError', 'message': 'Additional properties are not allowed (content was unexpected)'},
-    ]
+
+    assert len(response['errors']) == 1
+    assert response['errors'][0]['error'] == 'ValidationError'
+    # The error message for oneOf is ugly, non-deterministic in ordering
+    # and contains some UUID, so let’s just pick out the important bits
+    assert (
+        ' is valid under each of '
+    ) in response['errors'][0]['message']
+    assert (
+        '{required: [content]}'
+    ) in response['errors'][0]['message']
+    assert (
+        '{required: [template_id]}'
+    ) in response['errors'][0]['message']
+
+
+def test_create_broadcast_message_400s_if_reference_and_template_provided(
+    admin_request,
+    sample_broadcast_service,
+):
+    template = create_template(sample_broadcast_service, BROADCAST_TYPE)
+    response = admin_request.post(
+        'broadcast_message.create_broadcast_message',
+        _data={
+            'template_id': str(template.id),
+            'reference': 'abc123',
+            'service_id': str(sample_broadcast_service.id),
+            'created_by': str(sample_broadcast_service.created_by_id),
+        },
+        service_id=sample_broadcast_service.id,
+        _expected_status=400
+    )
+
+    assert len(response['errors']) == 1
+    assert response['errors'][0]['error'] == 'ValidationError'
+    # The error message for oneOf is ugly, non-deterministic in ordering
+    # and contains some UUID, so let’s just pick out the important bits
+    assert (
+        ' is valid under each of '
+    ) in response['errors'][0]['message']
+    assert (
+        '{required: [reference]}'
+    ) in response['errors'][0]['message']
+    assert (
+        '{required: [template_id]}'
+    ) in response['errors'][0]['message']
+
+
+def test_create_broadcast_message_400s_if_reference_not_provided_with_content(
+    admin_request,
+    sample_broadcast_service,
+):
+    response = admin_request.post(
+        'broadcast_message.create_broadcast_message',
+        _data={
+            'content': 'Some tailor made broadcast content',
+            'service_id': str(sample_broadcast_service.id),
+            'created_by': str(sample_broadcast_service.created_by_id),
+        },
+        service_id=sample_broadcast_service.id,
+        _expected_status=400
+    )
+    assert len(response['errors']) == 1
+    assert response['errors'][0]['error'] == 'ValidationError'
+    assert response['errors'][0]['message'].endswith(
+        'is not valid under any of the given schemas'
+    )
+
+
+def test_create_broadcast_message_400s_if_no_content_or_template(
+    admin_request,
+    sample_broadcast_service,
+):
+    response = admin_request.post(
+        'broadcast_message.create_broadcast_message',
+        _data={
+            'service_id': str(sample_broadcast_service.id),
+            'created_by': str(sample_broadcast_service.created_by_id),
+        },
+        service_id=sample_broadcast_service.id,
+        _expected_status=400
+    )
+    assert len(response['errors']) == 1
+    assert response['errors'][0]['error'] == 'ValidationError'
+    assert response['errors'][0]['message'].endswith(
+        'is not valid under any of the given schemas'
+    )
 
 
 @pytest.mark.parametrize('status', [
@@ -476,14 +579,17 @@ def test_update_broadcast_message_status_creates_event_with_correct_content_if_b
 
     assert alert_event.transmitted_content == {"body": "tailor made emergency broadcast content"}
 
-
+@pytest.mark.parametrize('is_platform_admin', [True, False])
 def test_update_broadcast_message_status_rejects_approval_from_creator(
     admin_request,
     sample_broadcast_service,
-    mocker
+    mocker,
+    is_platform_admin
 ):
     t = create_template(sample_broadcast_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
+    user = sample_broadcast_service.created_by
+    user.platform_admin = is_platform_admin
     mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
@@ -523,40 +629,6 @@ def test_update_broadcast_message_status_rejects_approval_of_broadcast_with_no_a
     ] == f'broadcast_message {broadcast.id} has no selected areas and so cannot be broadcasted.'
 
 
-def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_message(
-    notify_db,
-    admin_request,
-    sample_broadcast_service,
-    mocker
-):
-    user = sample_broadcast_service.created_by
-    user.platform_admin = True
-    t = create_template(sample_broadcast_service, BROADCAST_TYPE)
-    bm = create_broadcast_message(
-        t,
-        status=BroadcastStatusType.PENDING_APPROVAL,
-        areas={"areas": ["london"], "simple_polygons": [[[51.30, 0.7], [51.28, 0.8], [51.25, -0.7]]]}
-    )
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
-
-    response = admin_request.post(
-        'broadcast_message.update_broadcast_message_status',
-        _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(user.id)},
-        service_id=t.service_id,
-        broadcast_message_id=bm.id,
-        _expected_status=200
-    )
-
-    assert response['status'] == BroadcastStatusType.BROADCASTING
-    assert response['approved_at'] is not None
-    assert response['created_by_id'] == str(user.id)
-    assert response['approved_by_id'] == str(user.id)
-    mock_task.assert_called_once_with(
-        kwargs={'broadcast_event_id': str(bm.events[0].id)},
-        queue='broadcast-tasks'
-    )
-
-
 def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_own_message(
     notify_db,
     admin_request,

tests/app/celery/test_tasks.py

@@ -49,6 +49,7 @@ from app.models import (
     SMS_TYPE,
     ReturnedLetter,
     NOTIFICATION_CREATED)
+from app.serialised_models import SerialisedService, SerialisedTemplate
 from app.utils import DATETIME_FORMAT
 
 from tests.app import load_example_csv
@@ -1888,3 +1889,119 @@ def test_save_api_email_dont_retry_if_notification_already_exists(sample_service
     assert notifications[0].created_at == datetime(2020, 3, 25, 14, 30)
     # should only have sent the notification once.
     mock_provider_task.assert_called_once_with([data['id']], queue=expected_queue)
+
+
+@pytest.mark.parametrize('task_function, delivery_mock, recipient, template_args', (
+    (
+        save_email,
+        'app.celery.provider_tasks.deliver_email.apply_async',
+        'test@example.com',
+        {'template_type': 'email', 'subject': 'Hello'},
+    ), (
+        save_sms,
+        'app.celery.provider_tasks.deliver_sms.apply_async',
+        '07700 900890',
+        {'template_type': 'sms'}
+    ), (
+        save_letter,
+        'app.celery.letters_pdf_tasks.get_pdf_for_templated_letter.apply_async',
+        '123 Example Street\nCity of Town\nXM4 5HQ',
+        {'template_type': 'letter', 'subject': 'Hello'}
+    ),
+))
+def test_save_tasks_use_cached_service_and_template(
+    notify_db_session,
+    mocker,
+    task_function,
+    delivery_mock,
+    recipient,
+    template_args,
+):
+    service = create_service()
+    template = create_template(service=service, **template_args)
+
+    notification = _notification_json(template, to=recipient)
+    delivery_mock = mocker.patch(delivery_mock)
+    service_dict_mock = mocker.patch(
+        'app.serialised_models.SerialisedService.get_dict',
+        wraps=SerialisedService.get_dict,
+    )
+    template_dict_mock = mocker.patch(
+        'app.serialised_models.SerialisedTemplate.get_dict',
+        wraps=SerialisedTemplate.get_dict,
+    )
+
+    for _ in range(3):
+        task_function(
+            service.id,
+            uuid.uuid4(),
+            encryption.encrypt(notification),
+        )
+
+    # We talk to the database once for the service and once for the
+    # template; subsequent calls are caught by the in memory cache
+    assert service_dict_mock.call_args_list == [
+        call(service.id),
+    ]
+    assert template_dict_mock.call_args_list == [
+        call(str(template.id), str(service.id), 1),
+    ]
+
+    # But we save 3 notifications and enqueue 3 tasks
+    assert len(Notification.query.all()) == 3
+    assert len(delivery_mock.call_args_list) == 3
+
+
+@freeze_time('2020-03-25 14:30')
+@pytest.mark.parametrize('notification_type, task_function, expected_queue, recipient', (
+    ('sms', save_api_sms, QueueNames.SEND_SMS, '+447700900855'),
+    ('email', save_api_email, QueueNames.SEND_EMAIL, 'jane.citizen@example.com'),
+))
+def test_save_api_tasks_use_cache(
+    sample_service,
+    mocker,
+    notification_type,
+    task_function,
+    expected_queue,
+    recipient,
+):
+    mock_provider_task = mocker.patch(
+        f'app.celery.provider_tasks.deliver_{notification_type}.apply_async'
+    )
+    service_dict_mock = mocker.patch(
+        'app.serialised_models.SerialisedService.get_dict',
+        wraps=SerialisedService.get_dict,
+    )
+
+    template = create_template(sample_service, template_type=notification_type)
+    api_key = create_api_key(service=template.service)
+
+    def create_encrypted_notification():
+        return encryption.encrypt({
+            "to": recipient,
+            "id": str(uuid.uuid4()),
+            "template_id": str(template.id),
+            "template_version": template.version,
+            "service_id": str(template.service_id),
+            "personalisation": None,
+            "notification_type": template.template_type,
+            "api_key_id": str(api_key.id),
+            "key_type": api_key.key_type,
+            "client_reference": 'our email',
+            "reply_to_text": "our.email@gov.uk",
+            "document_download_count": 0,
+            "status": NOTIFICATION_CREATED,
+            "created_at": datetime.utcnow().strftime(DATETIME_FORMAT),
+        })
+
+    assert len(Notification.query.all()) == 0
+
+    for _ in range(3):
+        task_function(encrypted_notification=create_encrypted_notification())
+
+    assert service_dict_mock.call_args_list == [
+        call(str(template.service_id))
+    ]
+
+    assert len(Notification.query.all()) == 3
+    assert len(mock_provider_task.call_args_list) == 3

tests/app/clients/test_cbc_proxy.py

@@ -7,7 +7,7 @@ from unittest.mock import Mock, call
 import pytest
 
 from app.clients.cbc_proxy import (
-    CBCProxyClient, CBCProxyException, CBCProxyEE, CBCProxyCanary, CBCProxyVodafone, CBCProxyThree
+    CBCProxyClient, CBCProxyException, CBCProxyEE, CBCProxyCanary, CBCProxyVodafone, CBCProxyThree, CBCProxyO2
 )
 from app.utils import DATETIME_FORMAT
 
@@ -39,12 +39,6 @@ def cbc_proxy_client(client, mocker):
 def cbc_proxy_ee(cbc_proxy_client):
     return cbc_proxy_client.get_proxy('ee')
 
-
-@pytest.fixture
-def cbc_proxy_three(cbc_proxy_client):
-    return cbc_proxy_client.get_proxy('three')
-
-
 @pytest.fixture
 def cbc_proxy_vodafone(cbc_proxy_client):
     return cbc_proxy_client.get_proxy('vodafone')
@@ -53,6 +47,7 @@ def cbc_proxy_vodafone(cbc_proxy_client):
 @pytest.mark.parametrize('provider_name, expected_provider_class', [
     ('ee', CBCProxyEE),
     ('three', CBCProxyThree),
+    ('o2', CBCProxyO2),
     ('vodafone', CBCProxyVodafone),
     ('canary', CBCProxyCanary),
 ])
@@ -83,16 +78,15 @@ def test_cbc_proxy_lambda_client_has_correct_keys(cbc_proxy_ee):
     ('my-description', 'en-GB'),
     ('mŷ-description', 'cy-GB'),
 ))
-@pytest.mark.parametrize('cbc', ['bt-ee', 'three'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
 def test_cbc_proxy_one_2_many_create_and_send_invokes_function(
     mocker,
-    cbc_proxy_ee,
-    cbc_proxy_three,
+    cbc_proxy_client,
     description,
     cbc,
     expected_language,
 ):
-    cbc_proxy = cbc_proxy_ee if cbc == 'bt-ee' else cbc_proxy_three
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = 'my-identifier'
     headline = 'my-headline'
@@ -141,9 +135,9 @@ def test_cbc_proxy_one_2_many_create_and_send_invokes_function(
     assert payload['language'] == expected_language
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'three'])
-def test_cbc_proxy_one_2_many_cancel_invokes_function(mocker, cbc_proxy_ee, cbc_proxy_three, cbc):
-    cbc_proxy = cbc_proxy_ee if cbc == 'bt-ee' else cbc_proxy_three
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
+def test_cbc_proxy_one_2_many_cancel_invokes_function(mocker, cbc_proxy_client, cbc):
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = 'my-identifier'
     MockProviderMessage = namedtuple(
@@ -316,7 +310,7 @@ def test_cbc_proxy_vodafone_cancel_invokes_function(mocker, cbc_proxy_vodafone):
     assert payload['sent'] == sent
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_will_failover_to_second_lambda_if_function_error(
     mocker,
     cbc_proxy_client,
@@ -368,7 +362,7 @@ def test_cbc_proxy_will_failover_to_second_lambda_if_function_error(
     ]
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_will_failover_to_second_lambda_if_invoke_error(
     mocker,
     cbc_proxy_client,
@@ -415,7 +409,7 @@ def test_cbc_proxy_will_failover_to_second_lambda_if_invoke_error(
     ]
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_create_and_send_tries_failover_lambda_on_invoke_error_and_raises_if_both_invoke_error(
     mocker, cbc_proxy_client, cbc
 ):
@@ -458,7 +452,7 @@ def test_cbc_proxy_create_and_send_tries_failover_lambda_on_invoke_error_and_rai
     ]
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three'])
+@pytest.mark.parametrize('cbc', ['bt-ee', 'vodafone', 'three', 'o2'])
 def test_cbc_proxy_create_and_send_tries_failover_lambda_on_function_error_and_raises_if_both_function_error(
     mocker, cbc_proxy_client, cbc
 ):
@@ -538,9 +532,9 @@ def test_cbc_proxy_send_canary_invokes_function(mocker, cbc_proxy_client):
     assert payload['identifier'] == identifier
 
 
-@pytest.mark.parametrize('cbc', ['bt-ee', 'three'])
-def test_cbc_proxy_one_2_many_send_link_test_invokes_function(mocker, cbc_proxy_ee, cbc_proxy_three, cbc):
-    cbc_proxy = cbc_proxy_ee if cbc == 'bt-ee' else cbc_proxy_three
+@pytest.mark.parametrize('cbc', ['bt-ee', 'three', 'o2'])
+def test_cbc_proxy_one_2_many_send_link_test_invokes_function(mocker, cbc_proxy_client, cbc):
+    cbc_proxy = cbc_proxy_client.get_proxy('ee') if cbc == 'bt-ee' else cbc_proxy_client.get_proxy(cbc)
 
     identifier = str(uuid.uuid4())
 

tests/app/v2/notifications/test_post_notifications.py

@@ -233,7 +233,7 @@ def test_should_cache_template_lookups_in_memory(mocker, client, sample_template
 
     assert mock_get_template.call_count == 1
     assert mock_get_template.call_args_list == [
-        call(service_id=str(sample_template.service_id), template_id=str(sample_template.id))
+        call(service_id=str(sample_template.service_id), template_id=str(sample_template.id), version=None)
     ]
     assert Notification.query.count() == 5