Merge pull request #2944 from alphagov/send-events

Send events

app/broadcast_message/rest.py

@@ -3,18 +3,17 @@ from datetime import datetime
 import iso8601
 from flask import Blueprint, jsonify, request, current_app
 from app.config import QueueNames
+from app.dao.dao_utils import dao_save_object
 from app.dao.templates_dao import dao_get_template_by_id_and_service_id
 from app.dao.users_dao import get_user_by_id
 from app.dao.broadcast_message_dao import (
-    dao_create_broadcast_message,
     dao_get_broadcast_message_by_id_and_service_id,
     dao_get_broadcast_messages_for_service,
-    dao_update_broadcast_message,
 )
 from app.dao.services_dao import dao_fetch_service_by_id
 from app.errors import register_errors, InvalidRequest
-from app.models import BroadcastMessage, BroadcastStatusType
+from app.models import BroadcastMessage, BroadcastStatusType, BroadcastEvent, BroadcastEventMessageType
-from app.celery.broadcast_message_tasks import send_broadcast_message
+from app.celery.broadcast_message_tasks import send_broadcast_event
 from app.broadcast_message.broadcast_message_schema import (
     create_broadcast_message_schema,
     update_broadcast_message_schema,
@@ -109,7 +108,7 @@ def create_broadcast_message(service_id):
         created_by_id=user.id,
     )
 
-    dao_create_broadcast_message(broadcast_message)
+    dao_save_object(broadcast_message)
 
     return jsonify(broadcast_message.serialize()), 201
 
@@ -137,7 +136,7 @@ def update_broadcast_message(service_id, broadcast_message_id):
     if 'areas' in data:
         broadcast_message.areas = data['areas']
 
-    dao_update_broadcast_message(broadcast_message)
+    dao_save_object(broadcast_message)
 
     return jsonify(broadcast_message.serialize()), 200
 
@@ -153,12 +152,49 @@ def update_broadcast_message_status(service_id, broadcast_message_id):
     updating_user = get_user_by_id(data['created_by'])
 
     _update_broadcast_message(broadcast_message, new_status, updating_user)
-    dao_update_broadcast_message(broadcast_message)
+    dao_save_object(broadcast_message)
 
-    if new_status == BroadcastStatusType.BROADCASTING:
+    if new_status in {BroadcastStatusType.BROADCASTING, BroadcastStatusType.CANCELLED}:
-        send_broadcast_message.apply_async(
+        _create_broadcast_event(broadcast_message)
-            kwargs={'broadcast_message_id': str(broadcast_message.id)},
-            queue=QueueNames.NOTIFY
-        )
 
     return jsonify(broadcast_message.serialize()), 200
+
+
+def _create_broadcast_event(broadcast_message):
+    """
+    Creates a broadcast event, stores it in the database, and triggers the task to send the CAP XML off
+    """
+    msg_types = {
+        BroadcastStatusType.BROADCASTING: BroadcastEventMessageType.ALERT,
+        BroadcastStatusType.CANCELLED: BroadcastEventMessageType.CANCEL,
+    }
+
+    if broadcast_message.status == BroadcastStatusType.CANCELLED:
+        transmitted_finishes_at = broadcast_message.cancelled_at
+    else:
+        transmitted_finishes_at = broadcast_message.finishes_at
+
+    # TODO: This doesn't support placeholders yet. We shouldn't use BroadcastMessageTemplate when we add placeholders
+    # as that just outputs XML, we need the raw text.
+    event = BroadcastEvent(
+        service=broadcast_message.service,
+        broadcast_message=broadcast_message,
+        message_type=msg_types[broadcast_message.status],
+        transmitted_content={"body": broadcast_message.template.content},
+        transmitted_areas=broadcast_message.areas,
+        # TODO: Probably move this somewhere more standalone too and imply that it shouldn't change. Should it include
+        # a service based identifier too? eg "flood-warnings@notifications.service.gov.uk" or similar
+        transmitted_sender='notifications.service.gov.uk',
+
+        # TODO: Should this be set to now? Or the original starts_at?
+        transmitted_starts_at=broadcast_message.starts_at,
+        # TODO: When cancelling, do we need to set this to now? Or should we keep it as the original time.
+        transmitted_finishes_at=transmitted_finishes_at,
+    )
+
+    dao_save_object(event)
+
+    send_broadcast_event.apply_async(
+        kwargs={'broadcast_event_id': str(event.id)},
+        queue=QueueNames.NOTIFY
+    )

app/celery/broadcast_message_tasks.py

@@ -4,7 +4,7 @@ from notifications_utils.statsd_decorators import statsd
 
 from app import notify_celery
 
-from app.dao.broadcast_message_dao import dao_get_broadcast_message_by_id
+from app.dao.broadcast_message_dao import dao_get_broadcast_message_by_id, dao_get_broadcast_event_by_id
 
 
 @notify_celery.task(name="send-broadcast-message")
@@ -35,3 +35,27 @@ def send_broadcast_message(broadcast_message_id, provider='stub-1'):
         f'broadcast_message {broadcast_message.id} '
         f'status {broadcast_message.status} sent to {provider}'
     )
+
+
+@notify_celery.task(name="send-broadcast-event")
+@statsd(namespace="tasks")
+def send_broadcast_event(broadcast_event_id, provider='stub-1'):
+    broadcast_event = dao_get_broadcast_event_by_id(broadcast_event_id)
+
+    current_app.logger.info(
+        f'sending broadcast_event {broadcast_event.reference} '
+        f'msgType {broadcast_event.message_type} to {provider}'
+    )
+
+    payload = broadcast_event.serialize()
+
+    resp = requests.post(
+        f'{current_app.config["CBC_PROXY_URL"]}/broadcasts/events/{provider}',
+        json=payload
+    )
+    resp.raise_for_status()
+
+    current_app.logger.info(
+        f'broadcast_event {broadcast_event.reference} '
+        f'msgType {broadcast_event.message_type} sent to {provider}'
+    )

app/dao/broadcast_message_dao.py

@@ -1,16 +1,4 @@
-from app import db
 from app.models import BroadcastMessage, BroadcastEvent
-from app.dao.dao_utils import transactional
-
-
-@transactional
-def dao_create_broadcast_message(broadcast_message):
-    db.session.add(broadcast_message)
-
-
-@transactional
-def dao_update_broadcast_message(broadcast_message):
-    db.session.add(broadcast_message)
 
 
 def dao_get_broadcast_message_by_id_and_service_id(broadcast_message_id, service_id):
@@ -24,6 +12,10 @@ def dao_get_broadcast_message_by_id(broadcast_message_id):
     return BroadcastMessage.query.get(broadcast_message_id)
 
 
+def dao_get_broadcast_event_by_id(broadcast_event_id):
+    return BroadcastEvent.query.get(broadcast_event_id)
+
+
 def dao_get_broadcast_messages_for_service(service_id):
     return BroadcastMessage.query.filter(
         BroadcastMessage.service_id == service_id

app/dao/dao_utils.py

@@ -73,3 +73,9 @@ def version_class(*version_options):
 
 def dao_rollback():
     db.session.rollback()
+
+
+@transactional
+def dao_save_object(obj):
+    # add/update object in db
+    db.session.add(obj)

app/models.py

@@ -2332,21 +2332,25 @@ class BroadcastEvent(db.Model):
     transmitted_starts_at = db.Column(db.DateTime, nullable=True)
     transmitted_finishes_at = db.Column(db.DateTime, nullable=True)
 
-    # @property
+    @property
-    # def reference(self):
+    def reference(self):
-    #     # TODO: write this `from_event` function
+        return BroadcastMessageTemplate.from_event(self.serialize()).reference
-    #     return BroadcastMessageTemplate.from_event(self.serialize()).reference
+
+    def get_earlier_message_references(self):
+        from app.dao.broadcast_message_dao import get_earlier_events_for_broadcast_event
+        return [event.reference for event in get_earlier_events_for_broadcast_event(self.id)]
 
     def serialize(self):
         return {
-            'id': self.id,
+            'id': str(self.id),
 
-            'service_id': self.service_id,
+            'service_id': str(self.service_id),
 
-            # 'reference': self.reference,
+            'previous_event_references': self.get_earlier_message_references(),
 
-            'broadcast_message_id': self.broadcast_message_id,
+            'broadcast_message_id': str(self.broadcast_message_id),
-            'sent_at': self.sent_at,
+            # sent_at is required by BroadcastMessageTemplate.from_broadcast_event
+            'sent_at': self.sent_at.strftime(DATETIME_FORMAT),
             'message_type': self.message_type,
 
             'transmitted_content': self.transmitted_content,
@@ -2354,5 +2358,7 @@ class BroadcastEvent(db.Model):
             'transmitted_sender': self.transmitted_sender,
 
             'transmitted_starts_at': get_dt_string_or_none(self.transmitted_starts_at),
-            'transmitted_finishes_at': get_dt_string_or_none(self.transmitted_finishes_at),
+            # transmitted_finishes_at is required by BroadcastMessageTemplate.from_broadcast_event
+            'transmitted_finishes_at': self.transmitted_finishes_at.strftime(DATETIME_FORMAT),
+
         }

requirements-app.txt

@@ -29,7 +29,7 @@ notifications-python-client==5.5.1
 # PaaS
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@40.6.0#egg=notifications-utils==40.6.0
+git+https://github.com/alphagov/notifications-utils.git@40.9.0#egg=notifications-utils==40.9.0
 
 # gds-metrics requires prometheseus 0.2.0, override that requirement as 0.7.1 brings significant performance gains
 prometheus-client==0.7.1

requirements.txt

@@ -31,7 +31,7 @@ notifications-python-client==5.5.1
 # PaaS
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@40.6.0#egg=notifications-utils==40.6.0
+git+https://github.com/alphagov/notifications-utils.git@40.9.0#egg=notifications-utils==40.9.0
 
 # gds-metrics requires prometheseus 0.2.0, override that requirement as 0.7.1 brings significant performance gains
 prometheus-client==0.7.1
@@ -42,14 +42,14 @@ alembic==1.4.2
 amqp==1.4.9
 anyjson==0.3.3
 attrs==19.3.0
-awscli==1.18.109
+awscli==1.18.111
 bcrypt==3.1.7
 billiard==3.3.0.23
 bleach==3.1.4
 blinker==1.4
 boto==2.49.0
 boto3==1.10.38
-botocore==1.17.32
+botocore==1.17.34
 certifi==2020.6.20
 chardet==3.0.4
 click==7.1.2
@@ -62,6 +62,7 @@ geojson==2.5.0
 govuk-bank-holidays==0.6
 greenlet==0.4.16
 idna==2.10
+importlib-metadata==1.7.0
 Jinja2==2.11.2
 jmespath==0.10.0
 kombu==3.0.37
@@ -91,3 +92,4 @@ statsd==3.3.0
 urllib3==1.25.10
 webencodings==0.5.1
 Werkzeug==1.0.1
+zipp==3.1.0

tests/app/broadcast_message/test_rest.py

@@ -1,9 +1,10 @@
+from unittest.mock import ANY
 import uuid
 
 from freezegun import freeze_time
 import pytest
 
-from app.models import BROADCAST_TYPE, BroadcastStatusType
+from app.models import BROADCAST_TYPE, BroadcastStatusType, BroadcastEventMessageType
 
 from tests.app.db import create_broadcast_message, create_template, create_service, create_user
 
@@ -265,11 +266,12 @@ def test_update_broadcast_message_status_doesnt_let_you_update_other_things(admi
     }]
 
 
-def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(admin_request, sample_service):
+def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(admin_request, sample_service, mocker):
-    t = create_template(sample_service, BROADCAST_TYPE)
+    t = create_template(sample_service, BROADCAST_TYPE, content='emergency broadcast')
     bm = create_broadcast_message(t, status=BroadcastStatusType.BROADCASTING)
     canceller = create_user(email='canceller@gov.uk')
     sample_service.users.append(canceller)
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -279,21 +281,33 @@ def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(ad
         _expected_status=200
     )
 
+    assert len(bm.events) == 1
+    cancel_event = bm.events[0]
+
+    cancel_id = str(cancel_event.id)
+
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': cancel_id}, queue='notify-internal-tasks')
     assert response['status'] == BroadcastStatusType.CANCELLED
     assert response['cancelled_at'] is not None
     assert response['cancelled_by_id'] == str(canceller.id)
 
+    assert cancel_event.service_id == sample_service.id
+    assert cancel_event.transmitted_areas == bm.areas
+    assert cancel_event.message_type == BroadcastEventMessageType.CANCEL
+    assert cancel_event.transmitted_finishes_at == bm.cancelled_at
+    assert cancel_event.transmitted_content == {"body": "emergency broadcast"}
+
 
 def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_queues_task(
     admin_request,
     sample_service,
     mocker
 ):
-    t = create_template(sample_service, BROADCAST_TYPE)
+    t = create_template(sample_service, BROADCAST_TYPE, content='emergency broadcast')
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
     approver = create_user(email='approver@gov.uk')
     sample_service.users.append(approver)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -306,7 +320,17 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_
     assert response['status'] == BroadcastStatusType.BROADCASTING
     assert response['approved_at'] is not None
     assert response['approved_by_id'] == str(approver.id)
-    mock_task.assert_called_once_with(kwargs={'broadcast_message_id': str(bm.id)}, queue='notify-internal-tasks')
+
+    assert len(bm.events) == 1
+    alert_event = bm.events[0]
+
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': str(alert_event.id)}, queue='notify-internal-tasks')
+
+    assert alert_event.service_id == sample_service.id
+    assert alert_event.transmitted_areas == bm.areas
+    assert alert_event.message_type == BroadcastEventMessageType.ALERT
+    assert alert_event.transmitted_finishes_at == bm.finishes_at
+    assert alert_event.transmitted_content == {"body": "emergency broadcast"}
 
 
 def test_update_broadcast_message_status_rejects_approval_from_creator(
@@ -316,7 +340,7 @@ def test_update_broadcast_message_status_rejects_approval_from_creator(
 ):
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -340,7 +364,7 @@ def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_me
     user.platform_admin = True
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -354,7 +378,10 @@ def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_me
     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_message_id': str(bm.id)}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(
+        kwargs={'broadcast_event_id': str(bm.events[0].id)},
+        queue='notify-internal-tasks'
+    )
 
 
 def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_own_message(
@@ -366,7 +393,7 @@ def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_o
     sample_service.restricted = True
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -380,7 +407,7 @@ def test_update_broadcast_message_status_allows_trial_mode_services_to_approve_o
     assert response['approved_at'] is not None
     assert response['created_by_id'] == str(t.created_by_id)
     assert response['approved_by_id'] == str(t.created_by_id)
-    mock_task.assert_called_once_with(kwargs={'broadcast_message_id': str(bm.id)}, queue='notify-internal-tasks')
+    mock_task.assert_called_once_with(kwargs={'broadcast_event_id': ANY}, queue='notify-internal-tasks')
 
 
 def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_service(
@@ -391,7 +418,7 @@ def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_
     t = create_template(sample_service, BROADCAST_TYPE)
     bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL)
     approver = create_user(email='approver@gov.uk')
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',
@@ -423,7 +450,7 @@ def test_update_broadcast_message_status_restricts_status_transitions_to_explici
     bm = create_broadcast_message(t, status=current_status)
     approver = create_user(email='approver@gov.uk')
     sample_service.users.append(approver)
-    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async')
+    mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_event.apply_async')
 
     response = admin_request.post(
         'broadcast_message.update_broadcast_message_status',

tests/app/celery/test_broadcast_message_tasks.py

@@ -1,61 +1,118 @@
+from freezegun import freeze_time
 import pytest
 import requests_mock
 from requests import RequestException
 
 from app.dao.templates_dao import dao_update_template
-from app.models import BROADCAST_TYPE, BroadcastStatusType
+from app.models import BROADCAST_TYPE, BroadcastStatusType, BroadcastEventMessageType
-from app.celery.broadcast_message_tasks import send_broadcast_message
+from app.celery.broadcast_message_tasks import send_broadcast_message, send_broadcast_event
-from tests.app.db import create_template, create_broadcast_message
+from tests.app.db import create_template, create_broadcast_message, create_broadcast_event
 
 
 def test_send_broadcast_message_sends_data_correctly(sample_service):
-    t = create_template(sample_service, BROADCAST_TYPE)
+    template = create_template(sample_service, BROADCAST_TYPE)
-    bm = create_broadcast_message(t, areas=['london'], status=BroadcastStatusType.BROADCASTING)
+    broadcast_message = create_broadcast_message(template, areas=['london'], status=BroadcastStatusType.BROADCASTING)
 
     with requests_mock.Mocker() as request_mock:
         request_mock.post("http://test-cbc-proxy/broadcasts/stub-1", json={'valid': 'true'}, status_code=200)
-        send_broadcast_message(broadcast_message_id=str(bm.id))
+        send_broadcast_message(broadcast_message_id=str(broadcast_message.id))
 
     assert request_mock.call_count == 1
     assert request_mock.request_history[0].method == 'POST'
     assert request_mock.request_history[0].headers["Content-type"] == "application/json"
 
     cbc_json = request_mock.request_history[0].json()
-    assert cbc_json['template']['id'] == str(t.id)
+    assert cbc_json['template']['id'] == str(template.id)
     assert cbc_json['template']['template_type'] == BROADCAST_TYPE
     assert cbc_json['broadcast_message']['areas'] == ['london']
 
 
 def test_send_broadcast_message_sends_old_version_of_template(sample_service):
-    t = create_template(sample_service, BROADCAST_TYPE, content='first content')
+    template = create_template(sample_service, BROADCAST_TYPE, content='first content')
-    bm = create_broadcast_message(t, areas=['london'], status=BroadcastStatusType.BROADCASTING)
+    broadcast_message = create_broadcast_message(template, areas=['london'], status=BroadcastStatusType.BROADCASTING)
 
-    t.content = 'second content'
+    template.content = 'second content'
-    dao_update_template(t)
+    dao_update_template(template)
-    assert t.version == 2
+    assert template.version == 2
 
     with requests_mock.Mocker() as request_mock:
         request_mock.post("http://test-cbc-proxy/broadcasts/stub-1", json={'valid': 'true'}, status_code=200)
-        send_broadcast_message(broadcast_message_id=str(bm.id))
+        send_broadcast_message(broadcast_message_id=str(broadcast_message.id))
 
     assert request_mock.call_count == 1
     assert request_mock.request_history[0].method == 'POST'
     assert request_mock.request_history[0].headers["Content-type"] == "application/json"
 
     cbc_json = request_mock.request_history[0].json()
-    assert cbc_json['template']['id'] == str(t.id)
+    assert cbc_json['template']['id'] == str(template.id)
     assert cbc_json['template']['version'] == 1
     assert cbc_json['template']['content'] == 'first content'
 
 
 def test_send_broadcast_message_errors(sample_service):
-    t = create_template(sample_service, BROADCAST_TYPE)
+    template = create_template(sample_service, BROADCAST_TYPE)
-    bm = create_broadcast_message(t, status=BroadcastStatusType.BROADCASTING)
+    broadcast_message = create_broadcast_message(template, status=BroadcastStatusType.BROADCASTING)
 
     with requests_mock.Mocker() as request_mock:
         request_mock.post("http://test-cbc-proxy/broadcasts/stub-1", text='503 bad gateway', status_code=503)
         # we're not retrying or anything for the moment - but this'll ensure any exception gets logged
         with pytest.raises(RequestException) as ex:
-            send_broadcast_message(broadcast_message_id=str(bm.id))
+            send_broadcast_message(broadcast_message_id=str(broadcast_message.id))
+
+    assert ex.value.response.status_code == 503
+
+
+@freeze_time('2020-08-01 12:00')
+def test_send_broadcast_event_sends_data_correctly(sample_service):
+    template = create_template(sample_service, BROADCAST_TYPE)
+    broadcast_message = create_broadcast_message(template, areas=['london'], status=BroadcastStatusType.BROADCASTING)
+    event = create_broadcast_event(broadcast_message)
+
+    with requests_mock.Mocker() as request_mock:
+        request_mock.post("http://test-cbc-proxy/broadcasts/events/stub-1", json={'valid': 'true'}, status_code=200)
+        send_broadcast_event(broadcast_event_id=str(event.id))
+
+    assert request_mock.call_count == 1
+    assert request_mock.request_history[0].method == 'POST'
+    assert request_mock.request_history[0].headers["Content-type"] == "application/json"
+
+    cbc_json = request_mock.request_history[0].json()
+    assert cbc_json['id'] == str(event.id)
+    assert cbc_json['broadcast_message_id'] == str(broadcast_message.id)
+    assert cbc_json['sent_at'] == '2020-08-01T12:00:00.000000Z'
+    assert cbc_json['transmitted_starts_at'] is None
+    assert cbc_json['transmitted_areas'] == ['london']
+
+
+def test_send_broadcast_event_sends_references(sample_service):
+    template = create_template(sample_service, BROADCAST_TYPE, content='content')
+    broadcast_message = create_broadcast_message(template, areas=['london'], status=BroadcastStatusType.BROADCASTING)
+    alert_event = create_broadcast_event(broadcast_message, message_type=BroadcastEventMessageType.ALERT)
+    cancel_event = create_broadcast_event(broadcast_message, message_type=BroadcastEventMessageType.CANCEL)
+
+    with requests_mock.Mocker() as request_mock:
+        request_mock.post("http://test-cbc-proxy/broadcasts/events/stub-1", json={'valid': 'true'}, status_code=200)
+        send_broadcast_event(broadcast_event_id=str(cancel_event.id))
+
+    assert request_mock.call_count == 1
+    assert request_mock.request_history[0].method == 'POST'
+    assert request_mock.request_history[0].headers["Content-type"] == "application/json"
+
+    cbc_json = request_mock.request_history[0].json()
+    assert cbc_json['id'] == str(cancel_event.id)
+    assert cbc_json['message_type'] == cancel_event.message_type
+    assert cbc_json['previous_event_references'] == [alert_event.reference]
+
+
+def test_send_broadcast_event_errors(sample_service):
+    template = create_template(sample_service, BROADCAST_TYPE)
+    broadcast_message = create_broadcast_message(template, status=BroadcastStatusType.BROADCASTING)
+    event = create_broadcast_event(broadcast_message)
+
+    with requests_mock.Mocker() as request_mock:
+        request_mock.post("http://test-cbc-proxy/broadcasts/events/stub-1", text='503 bad gateway', status_code=503)
+        # we're not retrying or anything for the moment - but this'll ensure any exception gets logged
+        with pytest.raises(RequestException) as ex:
+            send_broadcast_event(broadcast_event_id=str(event.id))
 
     assert ex.value.response.status_code == 503

tests/app/db.py

@@ -1043,7 +1043,7 @@ def create_broadcast_event(
         transmitted_areas=transmitted_areas or ['london'],
         transmitted_sender=transmitted_sender or 'www.notifications.service.gov.uk',
         transmitted_starts_at=transmitted_starts_at,
-        transmitted_finishes_at=transmitted_finishes_at,
+        transmitted_finishes_at=transmitted_finishes_at or datetime.utcnow(),
     )
     db.session.add(b_e)
     db.session.commit()