add tests for get_earlier_events_for_broadcast_event

tests/app/dao/test_broadcast_message_dao.py

@@ -0,0 +1,43 @@
+from datetime import datetime
+from app.models import BROADCAST_TYPE
+from app.models import BroadcastEventMessageType
+from app.dao.broadcast_message_dao import get_earlier_events_for_broadcast_event
+
+from tests.app.db import create_broadcast_message, create_template, create_broadcast_event
+
+
+def test_get_earlier_events_for_broadcast_event(sample_service):
+    t = create_template(sample_service, BROADCAST_TYPE)
+    bm = create_broadcast_message(t)
+
+    events = [
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 12, 0, 0),
+            message_type=BroadcastEventMessageType.ALERT,
+            transmitted_content={'body': 'Initial content'}
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 13, 0, 0),
+            message_type=BroadcastEventMessageType.UPDATE,
+            transmitted_content={'body': 'Updated content'}
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 14, 0, 0),
+            message_type=BroadcastEventMessageType.UPDATE,
+            transmitted_content={'body': 'Updated content'},
+            transmitted_areas=['wales']
+        ),
+        create_broadcast_event(
+            bm,
+            sent_at=datetime(2020, 1, 1, 15, 0, 0),
+            message_type=BroadcastEventMessageType.CANCEL,
+            transmitted_finishes_at=datetime(2020, 1, 1, 15, 0, 0),
+        )
+    ]
+
+    # only fetches earlier events, and they're in time order
+    earlier_events = get_earlier_events_for_broadcast_event(events[2].id)
+    assert earlier_events == [events[0], events[1]]

tests/app/db.py

@@ -62,6 +62,7 @@ from app.models import (
     ServiceContactList,
     BroadcastMessage,
     BroadcastStatusType,
+    BroadcastEvent
 )
 
 
@@ -1021,3 +1022,29 @@ def create_broadcast_message(
     db.session.add(broadcast_message)
     db.session.commit()
     return broadcast_message
+
+
+def create_broadcast_event(
+    broadcast_message,
+    sent_at=None,
+    message_type='alert',
+    transmitted_content=None,
+    transmitted_areas=None,
+    transmitted_sender=None,
+    transmitted_starts_at=None,
+    transmitted_finishes_at=None,
+):
+    b_e = BroadcastEvent(
+        service=broadcast_message.service,
+        broadcast_message=broadcast_message,
+        sent_at=sent_at or datetime.utcnow(),
+        message_type=message_type,
+        transmitted_content=transmitted_content or {'body': 'this is an emergency broadcast message'},
+        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,
+    )
+    db.session.add(b_e)
+    db.session.commit()
+    return b_e